These nodes and sources connect legacy partners and systems that have no REST API. Credentials are never stored in the flow:
every secret is a single `${ENV}` or `${secret:name}` reference, and every host must be on
`transform.executor.http_host_allowlist` (the one egress allowlist of the connector).

| Kind | Name | What it does |
| --- | --- | --- |
| node | `soap` | SOAP 1.1/1.2 call, WS-Security, XML → JSON, Faults as failures |
| node | `sql` | Parameterized query on the configured Postgres |
| node | `email` | Send mail over SMTP with attachments |
| node | `sftp` | Write a file atomically |
| source | `sftp:{id}` | Poll a remote directory |
| source | `imap:{id}` | Poll a mailbox |
| capability | `mocks` | Stand-ins for systems that are not ready |

## SOAP node

```json
{ "type": "soap", "id": "get-order", "url": "https://erp.example.com/services/orders",
  "version": "1.1", "action": "urn:orders/GetOrder",
  "body": "<ns:GetOrder xmlns:ns=\"urn:orders\"><id>${order.id}</id><lines>${xml:lines}</lines></ns:GetOrder>",
  "wsse": { "username": "svc", "password": "${secret:erp_soap}", "password_type": "digest" } }
```

- It is an `http` node in SOAP mode: host allowlist, `auth`, posting `mode`, 401 refresh, dry run and `mock` all apply.
  A SOAP call is a POST, so `mode: read_only` skips it unless the node has `safe: true`.
- `${path}` values are XML-escaped. `${xml:path}` inserts a record field that holds a well-formed XML fragment as-is
  (build repeating items with a transform node).
- The response is parsed to JSON: attributes become `@attr`, repeated elements arrays, namespace prefixes are stripped
  (`response: {strip_prefixes, attr_prefix, text_key, arrays, typed, body_only}` to change that). With `merge: merge` the
  record is kept and the response lands in `metadata.soap_response`.
- A **SOAP Fault** (1.1 or 1.2, also inside HTTP 500) fails the node with `SOAP Fault code: string (detail)`;
  `on_fault: emit` continues with `soap_fault` on the record. HTTP 5xx/408/429 without a fault and transport errors are retried.
- **WSDL import**: in the node form, or `POST /api/soap/wsdl {wsdl | url}` / MCP `import_wsdl`: lists services, ports and
  operations (SOAPAction, endpoint, version) and generates a body skeleton from the schema (WSDL 1.1, inline schemas).

## SQL node

```json
{ "type": "sql", "id": "find", "query": "SELECT id, status FROM orders WHERE ref = ${order.ref}", "first": true, "read_only": true }
```

Every `${path}` becomes a bound parameter (`$1`, `$2`, …), never text in the statement: `'; DROP TABLE orders; --` is a
literal that matches nothing. No quotes around placeholders (deploy error), one statement per node, `${path?}` binds NULL
when the field is missing, `${secret:…}`/`${var:…}` bind too. `mode: execute` returns `{rows_affected}`. `read_only` runs a
`READ ONLY` transaction enforced by Postgres. Parameter types are inferred by the server; numeric, uuid, dates and json are
converted, anything else: `${x}::text::inet`. Needs the `postgres:` section.

## Email node

```json
{ "type": "email", "id": "recon-mail",
  "smtp": { "host": "smtp.example.com", "security": "starttls", "username": "${SMTP_USER}", "password": "${secret:smtp}" },
  "from": "Finance <bot@example.com>", "to": ["finance@example.com"], "subject": "Reconciliation ${day}",
  "body": "Report attached.", "html": "<p>Report for <b>${day}</b> attached.</p>",
  "attachments": [ { "filename": "recon-${day}.csv", "from": "reconcile.csv" } ] }
```

Text values are inserted as-is, HTML values escaped; subject and addresses reject line breaks; attachments are record
fields (a string or base64) up to 10 MB. 5xx replies and invalid addresses fail, connection errors and 4xx retry.

## SFTP

Built on `russh` (pure Rust, no libssh2/OpenSSL).

```json
{ "type": "source", "id": "orders-in", "match": "sftp:partner-orders",
  "sftp": { "host": "sftp.partner.example", "username": "opexflow", "private_key": "${secret:partner_key}",
        "host_key": "SHA256:…", "dir": "/outbound", "pattern": "ORD_*.csv", "format": "csv",
        "after": "archive", "archive_dir": "/outbound/done", "interval_secs": 120 } }
```

- **Host key pinning is required** (`host_key`: the `SHA256:…` fingerprint, a list during key rotation). A different key is
  refused; trust-on-first-use cannot be configured. `ssh-keyscan -t ed25519 host | ssh-keygen -lf -` prints it.
- **Source**: `format` csv, json, ndjson, xml, lines, fixed, raw. A processed-file ledger (`connectors.db`) keyed by path +
  size + mtime delivers every file once; the file is archived/deleted only after the flow settled all its records. A transient
  failure retries the whole file (max 5), so put a `dedupe` node on `_sftp.key`. Files younger than `min_age_secs` are left alone.
  `GET /api/connectors`, `GET /api/connectors/{node}/files`, `POST /api/connectors/{node}/poll`.
- **Node**: per record or per batch (`records: "rows"`), `json`/`ndjson`/`csv`/`text`, templated `dir`/`filename`,
  uploaded under a temporary name and renamed.

## IMAP source

`match: imap:{id}` with an `imap` object (`host`, `username`, `password`, `mailbox`, `search`, `attachments: metadata|content`).
Implicit TLS (or plain inside your network); messages are fetched with `BODY.PEEK` and flagged `\Seen` only after the flow
settled the record. Records carry `from`, `to`, `subject`, `date`, `headers`, `text`, `html`, `attachments` and `_imap`.

## Mock services (DX-06)

Define services in the flow's top-level `mocks` object (deployed and versioned with the flow) or in `<data_dir>/mocks.json`:

```json
{ "netsuite": { "endpoints": [
  { "method": "POST", "path": "/services/rest/record/v1/journalEntry",
"responses": [ { "status": 201, "body": { "id": "JE-${call}" } }, { "status": 409, "body": { "error": "duplicate" } } ],
"fault": { "kind": "status", "status": 503, "every": 5 } } ] } }
```

- Matching on method, path (`/a/{id}/b`, trailing `*`), query, headers and body (`json` subset, `contains`, `regex`); response
  sequences (`last` or `cycle`), delays, templates (`${request.body.x}`, `${request.path.id}`, `${call}`, `${uuid}`, `${now}`)
  and fault injection (`status`, `empty`, `malformed`, `timeout`; every n-th call or with a probability).
- Served at `/mock/{name}/…`. It needs the API token (operator) unless the service says `public: true`.
- `mock: true` (+ `mock_service` when several exist) on an `http`, `soap` or `sftp` node answers from the mock in-process, without the
  egress allowlist; a mocked sftp node/source works in `<data_dir>/mock/sftp/<name>`.
- **A protected environment refuses `mock` at deploy time and serves no mock.** Other environments get a deploy warning per mocked node.
- Every call is recorded (500 kept): **Mock services** in the top bar, `GET /api/mocks/calls`, MCP `mock_calls`.
- API: `GET /api/mocks`, `PUT|DELETE /api/mocks/{name}`, `POST /api/mocks/{name}/reset`.

## Limits

IMAP: no STARTTLS, public CA only. SFTP: no host certificates. sql: Postgres only, one statement. email: no inline images.
`fault` cannot reset a TCP connection. Binary size: SFTP and IMAP are the `sftp` and `imap` cargo features (default on).