flowforge-n8n — multi-tenant automation fabric

A control-plane / data-plane split for running automation workflows safely across tenants. n8n runs the workflows (the data plane); a FastAPI control plane owns everything security- and tenancy-sensitive: signed webhook ingress, per-tenant secret custody, replay protection, audit, and observability.

Proves: n8n · webhooks · FastAPI · multi-tenancy · secrets handling · observability.

flowchart LR
  Ext[External system] -->|signed webhook| CP[Control Plane: FastAPI]
  CP -->|verify HMAC + tenant| N8N[(n8n)]
  CP --> PG[(tenants, secrets, audit)]
  CP --> PROM[/metrics → Prometheus/]

Loading

Why this shape

The dependency arrow points one way: the control plane calls the data plane, never the reverse. n8n never sees an unauthenticated request and never holds a tenant's signing secret. That separation is the whole point — see docs/adr/0001-control-plane-data-plane.md.

Two decisions worth calling out:

1. Bought orchestration, built governance. n8n is a mature workflow engine; reinventing it would be wasteful. Tenancy, signed ingress, secret custody and audit are ours and live in code we test in isolation.
2. Ports & adapters for the scary parts. Secret storage is a port (SecretBox): Fernet locally, KMS/Vault in production — same interface, swap the adapter. The replay guard is an in-memory set here; in production it is a Redis SETNX. The core logic never changes.

Security guarantees (and the tests that hold them)

Every webhook to POST /hooks/{tenant} must pass three checks in signing.py:

Guarantee	How	Test
Authenticity	HMAC-SHA256 over "{ts}.".encode() + body, constant-time compare	test_valid_signature_passes, test_tampered_body_rejected
Freshness	reject timestamps outside a ±300s skew window	test_stale_timestamp_rejected
No replay	reject a signature already seen	test_replay_rejected
Tenant isolation	unknown tenant → 404, each tenant has its own secret	test_unknown_tenant_rejected
Secrets at rest	signing secrets are Fernet-sealed; plaintext never rests in the store	test_seal_open_roundtrip

Layout

src/flowforge/
├── config.py              # pydantic-settings (FLOWFORGE_* env vars)
├── logging.py             # structlog JSON logs
├── main.py                # composition root: control-plane HTTP API
└── adapters/
    ├── crypto.py          # SecretBox — Fernet seal/open (KMS/Vault stand-in)
    └── signing.py         # HMAC verify + timestamp freshness + replay guard
tests/{unit,integration}/  # security guarantees + end-to-end ingress
workflows/sample-workflow.json   # exported n8n workflow (reproducible import)
docs/adr/                  # architecture decisions

API

Method & path	Purpose
POST /tenants	register a tenant + signing secret (sealed at rest)
POST /hooks/{tenant}	signed webhook ingress → forwards to the tenant's n8n workflow
GET /audit	audit log of ingress attempts
GET /metrics	Prometheus metrics (workflow_runs_total, webhook_rejected_total)
GET /healthz	liveness

Interactive OpenAPI docs at /docs once running.

Run it

# generate a secrets-at-rest key (or: make key)
export FERNET_KEY=$(python -c "from cryptography.fernet import Fernet;print(Fernet.generate_key().decode())")
make up                        # n8n + its DB + control plane + Prometheus

PowerShell:

$env:FERNET_KEY = (uv run python -c "from cryptography.fernet import Fernet;print(Fernet.generate_key().decode())")
docker compose up -d --build

Demo — a valid call, a forged call, the audit trail

TS=$(python -c "import time;print(time.time())")
BODY='{"event":"deal.won","value":5000}'
SIG=$(python -c "import hmac,hashlib;print(hmac.new(b'tenant-acme-signing-secret', f'$TS.'.encode()+b'$BODY', hashlib.sha256).hexdigest())")

curl -s localhost:8000/hooks/acme -H "x-signature: $SIG" -H "x-timestamp: $TS" -d "$BODY"     # → {"status":"triggered"}
curl -s localhost:8000/hooks/acme -H "x-signature: deadbeef" -H "x-timestamp: $TS" -d "$BODY" # → 401, audited
curl -s localhost:8000/audit       # both attempts recorded
curl -s localhost:8000/metrics     # workflow_runs_total / webhook_rejected_total

Import the sample workflow into n8n (UI at localhost:5678) from workflows/sample-workflow.json, or re-export your own:

docker compose exec n8n n8n export:workflow --all --output=/tmp/wf.json
docker compose cp n8n:/tmp/wf.json workflows/sample-workflow.json

Develop

uv sync --all-extras --dev
make test        # pytest
make lint        # ruff check + format check
make typecheck   # mypy
uv run pre-commit install

CI runs lint + mypy + tests on every push — see .github/workflows/ci.yml.

Configuration

All settings are FLOWFORGE_-prefixed env vars (see .env.example):

Var	Default	Meaning
FLOWFORGE_N8N_URL	http://n8n:5678	data-plane base URL
FLOWFORGE_FERNET_KEY	dev key	secrets-at-rest key (override in prod)
FLOWFORGE_SIGNATURE_SKEW_S	300	replay window in seconds
FLOWFORGE_LOG_LEVEL	INFO	log level

The default Fernet key is a throwaway for local dev. Generate a real one with make key and never commit it.