Expanso: attest where the data is read and where the result leaves.

Where Makoto sits

Expanso Edge’s command processor runs a program per message with the message on stdin and replaces the message with its stdout. That is exactly the seam a seal needs.

1. batched inputReads sensor lines and groups them into batches of up to 500 or 30 seconds.
2. Normalize and archiveA Bloblang mapping normalizes each reading; archive: lines joins the batch into one message.
3. command sealWrites the batch to the outbox, signs an origin statement with the site’s key, and emits a receipt.
4. switch outputReceipts go to the receipts log. An errored message goes to unsealed/ and is never forwarded as evidence.
5. CollectorCentrally, checks each envelope against its receipt, hands off the batch, and verifies before admitting it.

1. Install on each edge node

Every platform on this site uses the same two files from the integration index: makoto_boundary.py, a standard-library helper that signs, hands off, and verifies through the pinned CLI, and makoto_bootstrap.py, which creates one key per claim role and the receiver’s trust policy. The edge node needs only the core checkout, jq, and its origin key; the collector needs the helper and the policy.

git clone https://github.com/makoto-project/makoto.git /opt/makoto
git -C /opt/makoto checkout 276008eeec0f10d5322c4c627b8c103222fee841
(cd /opt/makoto && uv sync --locked --dev)
export MAKOTO_CORE=/opt/makoto
export MAKOTO_KEYS=/etc/makoto/keys MAKOTO_OUTBOX=/var/lib/makoto/outbox SITE_ID=plant-7

uv run makoto_bootstrap.py /srv/makoto \
  --source-kind urn:example:source:edge-sensor-feed \
  --source-uri urn:example:site:plant-7 \
  --operation-type urn:example:operation:none
export MAKOTO_KEYS=/srv/makoto/keys MAKOTO_POLICY=/srv/makoto/policy.json

2. The pipeline

input:
  batched:
    child:
      file:
        paths: [ "/var/lib/sensors/readings-*.ndjson" ]
        scanner:
          lines: {}
    policy:
      count: 500
      period: 30s
      processors:
        # Normalize each reading, then join the batch into one NDJSON message.
        - mapping: |
            root = this
            root.site = env("SITE_ID")
            root.celsius = ((this.fahrenheit - 32) * 5 / 9).round()
            root.fahrenheit = deleted()
        - archive:
            format: lines

pipeline:
  processors:
    # Seal: write the batch, sign it, emit a receipt. A non-zero exit marks the
    # message as errored instead of stopping the pipeline.
    - command:
        name: bash
        args_mapping: '[ "/opt/makoto-edge/seal-batch.sh" ]'

output:
  switch:
    cases:
      # Never forward an unsealed batch as if it were evidence.
      - check: errored()
        output:
          file:
            path: /var/lib/makoto/unsealed/${! timestamp_unix_nano() }.ndjson
            codec: lines
      - output:
          file:
            path: /var/lib/makoto/receipts.ndjson
            codec: lines

expanso-edge validate pipeline.yaml
[OK] pipeline.yaml: valid

3. The seal

Install as /opt/makoto-edge/seal-batch.sh.

#!/usr/bin/env bash
# Expanso Edge `command` processor target: one archived batch arrives on stdin.
# Writes the batch, signs an origin statement over its exact bytes, and prints
# a one-line JSON receipt that replaces the message. Diagnostics go to stderr.
set -euo pipefail
: "${MAKOTO_CORE:?}" "${MAKOTO_KEYS:?}" "${MAKOTO_OUTBOX:?}" "${SITE_ID:?}"

makoto() { (cd "$MAKOTO_CORE" && uv run --quiet makoto "$@"); }

batch="$(date -u +%Y%m%dT%H%M%SZ)-$$-$RANDOM"
dir="$MAKOTO_OUTBOX/$batch"
mkdir -p "$dir"
cat > "$dir/readings.ndjson"

makoto attest origin \
  --subject "readings.ndjson=$dir/readings.ndjson" \
  --source-kind urn:example:source:edge-sensor-feed \
  --source-uri "urn:example:site:$SITE_ID" \
  --key "$MAKOTO_KEYS/origin.private.pem" \
  --out "$dir/origin.dsse.json" >&2

jq -cn \
  --arg batch "$batch" \
  --arg site "$SITE_ID" \
  --arg artifact "$(makoto digest "$dir/readings.ndjson")" \
  --argjson envelope "$(makoto envelope inspect --envelope "$dir/origin.dsse.json" --json)" \
  '{batch: $batch, site: $site, artifact: $artifact,
    statement: ("sha256:" + $envelope.payloadDigest.sha256),
    signedBy: $envelope.signatureKeyIds}'

Run by hand with a two-reading batch, it prints one receipt:

{"batch":"20260916T125702Z-98483-7858","site":"plant-7","artifact":"sha256:42722cbab08ac95edcec1923024ff64259a159b7da9b080cb14963b9ce645d56","statement":"sha256:bbad33a09eb10b6c81ecae200aa44e5a65e97ac8d12aea305e303638bd422711","signedBy":["sha256:b5a465bf91f8e5785d0f8a3abf65ede1338159e6794b5600c6d629948e269620"]}

4. The collector

Save this as collect_batches.py beside makoto_boundary.py.

"""Collector: hand off each sealed edge batch and admit it only if Makoto allows.

    python collect_batches.py /var/lib/makoto/receipts.ndjson /var/lib/makoto/outbox
"""

import json
import shutil
import sys
import uuid
from pathlib import Path

import makoto_boundary as mb

RECIPIENT = "example:central-lake"


def admit(receipt: dict, outbox: Path, tamper: bool = False) -> bool:
    batch = outbox / receipt["batch"]
    origin, readings = batch / "origin.dsse.json", batch / "readings.ndjson"
    # The receipt travelled separately from the files: it is the independent anchor.
    if "sha256:" + mb.statement_digest(origin) != receipt["statement"]:
        print(f"REFUSED {receipt['batch']}: envelope does not match its receipt", file=sys.stderr)
        return False
    anchors = mb.create_handoff(
        origin, readings, batch, batch / "bundle",
        recipient=RECIPIENT,
        bundle_id=f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, receipt['batch'])}",
        media_type="application/x-ndjson",
    )
    anchors["artifact"]["digest"]["sha256"] = receipt["artifact"].removeprefix("sha256:")
    delivered = batch / "delivered"
    shutil.copytree(batch / "bundle", delivered, dirs_exist_ok=True)
    if tamper:
        final = next((delivered / "artifacts" / "final").glob("*.bin"))
        final.write_bytes(final.read_bytes().replace(b'"celsius":21', b'"celsius":12'))
    try:
        mb.verify(delivered, anchors, recipient=RECIPIENT, report_path=batch / "verification.json")
    except mb.MakotoDenied as denied:
        print(f"REFUSED {receipt['batch']} from {receipt['site']}: {denied}", file=sys.stderr)
        return False
    print(f"ADMITTED {receipt['batch']} from {receipt['site']}")
    return True


if __name__ == "__main__":
    receipts, outbox = Path(sys.argv[1]), Path(sys.argv[2])
    tamper = "--tamper-in-transit" in sys.argv
    results = [admit(json.loads(line), outbox, tamper) for line in receipts.read_text().splitlines() if line]
    sys.exit(0 if all(results) else 1)

python collect_batches.py /var/lib/makoto/receipts.ndjson /var/lib/makoto/outbox
ADMITTED 20260916T125702Z-98483-7858 from plant-7

5. What a gate failure looks like

Change one reading in the delivered batch, as a compromised relay would:

python collect_batches.py receipts.ndjson outbox --tamper-in-transit
REFUSED 20260916T125702Z-98483-7858 from plant-7: Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest

Design notes

Why sign at the edge

A signature made after the batch reaches the cloud proves only what the cloud received. Signing on the node that read the sensor is what lets the collector detect a change in transit.

One key per site

Give each edge node its own origin key and list it in the receiver’s policy with that site’s source URI. A stolen key then speaks only for one site.

Error handling

The command processor marks a message as errored on a non-zero exit and lets it continue. Without the errored() case, an unsealed batch would be written where receipts belong.

Other attachment points on this platform

Local source read

Create or resolve an origin statement for the exact local artifact bytes before processing.

Job completion

After output commits, append a transformation statement with job, node, image, query, and execution context as operation metadata.

Distributed result

Use a dataset manifest when one logical result spans multiple nodes or files; preserve each exact member digest.

Outbound handoff

Sign the exact artifacts and evidence leaving the site; the receiver applies its own trust policy and profiles.

What the evidence can say

It can bind a data-local result to exact inputs, an operation claim, and the next handoff. It cannot prove HIPAA compliance, guarantee the claimed computation ran, or make location metadata confidential.

Troubleshooting

[FAIL] pipeline.yaml: ... bloblang syntax error

expanso-edge validate checks mappings offline. Run it in CI before deploying a pipeline through Expanso Cloud.

Batches land in unsealed/

The seal exited non-zero. Its stderr is attached to the errored message; the usual causes are an unset SITE_ID, uv missing from the agent’s PATH, or an unreadable key.

REFUSED ...: envelope does not match its receipt

The envelope in the outbox is not the one the edge node reported. Treat it as tampering, not as a transient error.