Airflow: attest after the task commits its output.

Where Makoto sits in the DAG

Each data-producing task is followed by a task that signs what it just committed. Signing is a separate task on purpose: the transformation can be retried or rewritten without ever touching key material, and a failed signature is visible in the graph as its own red box.

1. extractCommits the raw export for this run to a run-scoped directory.
2. attest_originSigns an origin statement over the exact raw bytes with the origin key, naming the source kind and URI the receiver’s policy trusts.
3. normalizeYour transformation. It knows nothing about Makoto.
4. attest_normalizeSigns a transformation statement that binds the output to the origin statement and the exact input digest, with the normalize key.
5. create_handoffPackages the statements and final artifact into a bundle signed by the handoff key, addressed to one recipient, and computes the three anchors the receiver will check.
6. deliverMoves the bundle to the receiver. In this demonstration a DAG parameter can corrupt it on the way.
7. makoto_gateRuns the verifier with the receiver’s policy and independent anchors. A denial raises AirflowFailException, so nothing downstream runs and nothing is retried.
8. publishRuns only on an allow, and records the manifest and policy digests the decision was made under.

1. Install the verifier beside Airflow

The workers call the reference CLI, so the pinned core checkout and uv must exist on every machine that runs a Makoto task. Airflow and Makoto keep separate Python environments; nothing is imported across them.

# The reference verifier, pinned and synced once per worker image.
git clone https://github.com/makoto-project/makoto.git /opt/makoto
git -C /opt/makoto checkout 276008eeec0f10d5322c4c627b8c103222fee841
(cd /opt/makoto && uv sync --locked --dev)

# Airflow itself, installed with its published constraints.
uv pip install "apache-airflow==3.3.1" \
  --constraint https://raw.githubusercontent.com/apache/airflow/constraints-3.3.1/constraints-3.12.txt

# Every scheduler, DAG processor, and worker needs both variables.
export MAKOTO_CORE=/opt/makoto
export MAKOTO_WORK=/srv/makoto

2. Create role keys and the receiver policy

Run this once per environment, outside Airflow. It creates one Ed25519 key pair per claim role and writes the receiver’s trust policy: which key may sign origins for which source, which key may sign which operation, and what the handoff must prove. The last line runs makoto policy check, so a malformed policy fails here rather than inside a task.

"""One-time setup: signing keys per claim role and the receiver's trust policy.

Run once per environment, outside Airflow:
    MAKOTO_CORE=/path/to/makoto MAKOTO_WORK=/srv/makoto uv run bootstrap_makoto.py
"""

import base64
import json
import os
import subprocess
from pathlib import Path

CORE = Path(os.environ["MAKOTO_CORE"]).resolve()
WORK = Path(os.environ["MAKOTO_WORK"]).resolve()
ROLES = ("origin", "normalize", "handoff")


def makoto(*args: object) -> str:
    result = subprocess.run(
        ["uv", "run", "--quiet", "makoto", *map(str, args)],
        cwd=CORE, capture_output=True, text=True, check=False,
    )
    if result.returncode != 0:
        raise SystemExit(result.stderr.strip())
    return result.stdout


def public_key_entry(role: str) -> tuple[str, dict]:
    public_pem = WORK / "keys" / f"{role}.public.pem"
    keyid = json.loads(makoto("key", "inspect", "--public", public_pem, "--json"))["keyid"]
    # The policy stores the DER SubjectPublicKeyInfo, base64-encoded: the PEM body.
    der = base64.b64decode("".join(public_pem.read_text().strip().splitlines()[1:-1]))
    return keyid, {"type": "ed25519", "publicKey": base64.b64encode(der).decode(), "label": f"orders pipeline {role} key"}


(WORK / "keys").mkdir(parents=True, exist_ok=True)
(WORK / "receiver").mkdir(parents=True, exist_ok=True)
for role in ROLES:
    if not (WORK / "keys" / f"{role}.private.pem").exists():
        makoto(
            "key", "generate",
            "--private-out", WORK / "keys" / f"{role}.private.pem",
            "--public-out", WORK / "keys" / f"{role}.public.pem",
        )

keys = dict(public_key_entry(role) for role in ROLES)
keyid = {role: kid for role, kid in zip(ROLES, keys)}

policy = {
    "version": "0.2",
    "keys": keys,
    "rules": [
        {
            "id": "urn:example:rule:orders-origin",
            "predicateTypes": ["https://usemakoto.dev/predicate/v0.2/origin"],
            "authorizedKeyIds": [keyid["origin"]],
            "minimumSignatures": 1,
            "sourceKinds": ["urn:example:source:orders-export"],
            "sourceUris": ["urn:example:orders:daily"],
        },
        {
            "id": "urn:example:rule:orders-normalize",
            "predicateTypes": ["https://usemakoto.dev/predicate/v0.2/transform"],
            "authorizedKeyIds": [keyid["normalize"]],
            "minimumSignatures": 1,
            "operationTypes": ["urn:example:operation:normalize-orders"],
        },
    ],
    "handoff": {
        "authorizedKeyIds": [keyid["handoff"]],
        "minimumSignatures": 1,
        "requireExpectedManifest": True,
        "requireExpectedHead": True,
        "requireExpectedArtifacts": True,
        "requireRecipient": True,
        "requireNonce": False,
        "allowReplayableHandoff": False,
        "maxAgeSeconds": 86400,
        "maxFutureSkewSeconds": 300,
    },
    "requiredProfiles": [],
    "limits": {
        "maxBundleFiles": 10000,
        "maxMetadataBytes": 104857600,
        "maxArtifactBytesPerFile": 10737418240,
        "maxAggregateArtifactBytes": 53687091200,
        "maxSnapshotBytes": 53687091200,
        "maxArtifactValidationBytes": 104857600,
        "maxJsonDepth": 128,
        "maxJsonNumberChars": 1024,
        "maxJsonExponentMagnitude": 10000,
        "maxSchemaBytes": 2097152,
        "maxSchemaResources": 256,
        "maxSchemaEvaluationDepth": 256,
        "maxSchemaOperations": 10000000,
        "maxRegexLength": 4096,
        "profileEvaluationTimeoutSeconds": 5,
        "profileWorkerMemoryBytes": 536870912,
        "maxNdjsonLineBytes": 1048576,
        "maxSignaturesTotal": 10000,
        "maxProfileEvaluations": 10000,
        "maxDiagnostics": 10000,
        "maxReportRecords": 20000,
        "maxReportBytes": 67108864,
    },
}
# The verifier requires canonical order: rules sorted by id, compared as bytes.
policy["rules"].sort(key=lambda rule: rule["id"].encode())
policy_path = WORK / "receiver" / "policy.json"
policy_path.write_text(json.dumps(policy, indent=2) + "\n", encoding="utf-8")
makoto("policy", "check", "--policy", policy_path)
print(f"keys in {WORK / 'keys'}; policy checked at {policy_path}")

MAKOTO_CORE=/opt/makoto MAKOTO_WORK=/srv/makoto uv run bootstrap_makoto.py
keys in /srv/makoto/keys; policy checked at /srv/makoto/receiver/policy.json

In a real deployment these are two parties. The producer holds the private keys. The receiver owns policy.json and needs only the public keys. The script creates both halves in one place so the example runs on one machine.

3. The DAG

Save this as dags/makoto_orders.py. It uses only the Airflow Task SDK and the standard library.

"""Attest every task boundary with Makoto, hand off a bundle, and gate on the verdict."""

from __future__ import annotations

import base64
import hashlib
import json
import os
import re
import shutil
import subprocess
import uuid
from pathlib import Path

import pendulum
from airflow.sdk import Param, dag, get_current_context, task
from airflow.sdk.exceptions import AirflowFailException

CORE = Path(os.environ["MAKOTO_CORE"])  # reviewed core checkout, synced with uv
WORK = Path(os.environ["MAKOTO_WORK"])  # keys/ and receiver/ from bootstrap_makoto.py
RECIPIENT = "example:analytics-team"
NORMALIZE_OP = "urn:example:operation:normalize-orders"


# --------------------------------------------------------------------------- helpers
def makoto(*args: object) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["uv", "run", "--quiet", "makoto", *map(str, args)],
        cwd=CORE, capture_output=True, text=True, check=False,
    )


def run_dir(*parts: str) -> Path:
    """One directory per DAG run, so retries and reruns never mix evidence."""
    run_id = re.sub(r"[^A-Za-z0-9._-]", "_", get_current_context()["run_id"])
    path = WORK.joinpath("runs", run_id, *parts)
    path.mkdir(parents=True, exist_ok=True)
    return path


def event_id(name: str) -> str:
    """Stable across task retries, unique across runs."""
    context = get_current_context()
    return f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{context['ti'].dag_id}/{context['run_id']}/{name}')}"


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def payload(envelope: Path) -> bytes:
    return base64.b64decode(json.loads(envelope.read_text())["payload"])


def sign(kind: str, role: str, subject: Path, out: Path, *extra: object) -> str:
    """Sign once. A retry reuses the envelope only if it covers the same bytes."""
    if out.exists():
        signed = json.loads(payload(out))["subject"][0]["digest"]["sha256"]
        if signed == sha256(subject):
            return str(out)
        raise AirflowFailException(f"{out.name} already attests different bytes; refusing a second history")
    result = makoto(
        "attest", kind,
        "--subject", f"{subject.name}={subject}",
        *extra,
        "--event-id", event_id(out.name),
        "--key", WORK / "keys" / f"{role}.private.pem",
        "--out", out,
    )
    if result.returncode != 0:
        raise RuntimeError(f"makoto attest {kind} failed: {result.stderr.strip()}")  # retryable
    return str(out)


# --------------------------------------------------------------------------- DAG
@dag(
    dag_id="makoto_orders",
    schedule=None,
    start_date=pendulum.datetime(2026, 9, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 2},
    params={"tamper_in_transit": Param(False, type="boolean")},
    tags=["makoto"],
)
def makoto_orders():
    @task
    def extract() -> str:
        raw = run_dir("data") / "orders.raw.csv"
        raw.write_text("order_id,region,total\n1001, us ,42.00\n1002,GB,17.50\n")
        return str(raw)

    @task
    def attest_origin(raw: str) -> str:
        return sign(
            "origin", "origin", Path(raw), run_dir("attestations") / "01-origin.dsse.json",
            "--source-kind", "urn:example:source:orders-export",
            "--source-uri", "urn:example:orders:daily",
        )

    @task
    def normalize(raw: str) -> str:
        header, *rows = Path(raw).read_text().splitlines()
        out = run_dir("data") / "orders.normalized.csv"
        cleaned = [",".join(cell.strip().lower() for cell in row.split(",")) for row in rows]
        out.write_text("\n".join([header, *cleaned]) + "\n")
        return str(out)

    @task
    def attest_normalize(raw: str, origin: str, normalized: str) -> str:
        binding = run_dir("bindings") / "normalize-input.json"
        binding.write_text(json.dumps({
            "name": "raw",
            "path": raw,
            "predecessor": origin,
            "subjectName": Path(raw).name,
        }))
        return sign(
            "transform", "normalize", Path(normalized),
            run_dir("attestations") / "02-normalize.dsse.json",
            "--input-binding", binding,
            "--operation-type", NORMALIZE_OP,
            "--operation-name", "Normalize order regions",
        )

    @task
    def create_handoff(head: str, normalized: str) -> dict:
        binding = run_dir("bindings") / "final-artifact.json"
        binding.write_text(json.dumps({
            "head": head,
            "subjectName": Path(normalized).name,
            "path": normalized,
            "mediaType": "text/csv",
        }))
        bundle = run_dir() / "bundle"
        result = makoto(
            "handoff", "create",
            "--head", head,
            "--attestations", run_dir("attestations"),
            "--artifact-binding", binding,
            "--recipient", RECIPIENT,
            "--bundle-id", event_id("bundle"),
            "--key", WORK / "keys" / "handoff.private.pem",
            "--out", bundle,
            "--force",  # a retry rebuilds the same bundle from the same signed inputs
        )
        if result.returncode != 0:
            raise RuntimeError(f"makoto handoff create failed: {result.stderr.strip()}")
        index = json.loads((bundle / "bundle.json").read_text())
        # These three anchors travel to the receiver out of band: a ticket, a
        # release record, a catalog entry. Never inside the bundle they describe.
        return {
            "bundle": str(bundle),
            "manifest": "sha256:" + hashlib.sha256(payload(bundle / index["manifest"])).hexdigest(),
            "head": "sha256:" + hashlib.sha256(payload(Path(head))).hexdigest(),
            "artifact": {
                "head": {"sha256": hashlib.sha256(payload(Path(head))).hexdigest()},
                "subjectName": Path(normalized).name,
                "digest": {"sha256": sha256(Path(normalized))},
            },
        }

    @task
    def deliver(handoff: dict) -> str:
        inbox = run_dir("receiver-inbox") / "bundle"
        shutil.copytree(handoff["bundle"], inbox, dirs_exist_ok=True)
        if get_current_context()["params"]["tamper_in_transit"]:
            final = next((inbox / "artifacts" / "final").glob("*.bin"))
            final.write_bytes(final.read_bytes() + b"1003,us,0.01\n")
        return str(inbox)

    @task
    def makoto_gate(delivered: str, handoff: dict) -> dict:
        expected_artifact = run_dir("receiver-inbox") / "expected-artifact.json"
        expected_artifact.write_text(json.dumps(handoff["artifact"]))
        result = makoto(
            "verify", "bundle", delivered,
            "--policy", WORK / "receiver" / "policy.json",
            "--expected-manifest", handoff["manifest"],
            "--expected-head", handoff["head"],
            "--expected-artifact", expected_artifact,
            "--expected-recipient", RECIPIENT,
            "--json",
        )
        if result.returncode not in (0, 1):
            raise RuntimeError(f"verifier could not run: {result.stderr.strip()}")  # retryable
        report_path = run_dir("reports") / "verification.json"
        report_path.write_text(result.stdout)
        report = json.loads(result.stdout)
        if report["decision"] != "allow":
            failed = [f"step {e['step']} {e['code']}: {e['message']}" for e in report["errors"]]
            # AirflowFailException skips retries: a denial will not change on retry.
            raise AirflowFailException(
                f"Makoto {report['decision']} ({report['primaryError']}); "
                f"report at {report_path}; " + "; ".join(failed)
            )
        return {
            "decision": report["decision"],
            "manifest": report["manifestDigest"]["sha256"],
            "policy": report["policyDigest"]["sha256"],
            "report": str(report_path),
        }

    @task
    def publish(verdict: dict, normalized: str) -> None:
        print(f"publishing {normalized} under Makoto {verdict['decision']} "
              f"(manifest {verdict['manifest'][:12]}, policy {verdict['policy'][:12]})")

    raw = extract()
    origin = attest_origin(raw)
    normalized = normalize(raw)
    head = attest_normalize(raw, origin, normalized)
    handoff = create_handoff(head, normalized)
    verdict = makoto_gate(deliver(handoff), handoff)
    publish(verdict, normalized)


makoto_orders()

4. Run it

airflow db migrate
airflow dags test makoto_orders

All eight tasks succeed. The publish task logs the decision and the digests it was made under:

publishing /srv/makoto/runs/manual__2026-09-16T12_46_35.467633_00_00/data/orders.normalized.csv under Makoto allow (manifest 712954fd6b0d, policy b0b749267a9b)

5. What a gate failure looks like

Trigger the same DAG with the data corrupted between handoff and verification. Everything upstream still succeeds, because every signature was honest when it was made. The gate is where the change is caught.

airflow dags test makoto_orders -c '{"tamper_in_transit": true}'

airflow.sdk.exceptions.AirflowFailException: Makoto deny (E_ARTIFACT_DIGEST); report at /srv/makoto/runs/manual__2026-09-16T12_46_45.122670_00_00/reports/verification.json; step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest

Task states for that run. makoto_gate fails once with no retries, and publish never starts:

task_id state
extract success
attest_origin success
normalize success
attest_normalize success
create_handoff success
deliver success
makoto_gate failed
publish upstream_failed

The report written by the gate says precisely what failed and what still held. Signatures, authorization, the graph, and the recipient all pass; only the delivered bytes do not match:

jq '{decision, primaryError,
    notPassed: [.checks[] | select(.status != "pass") | {id, status}],
    errors: [.errors[] | {step, code, message}],
    recipient: .actualRecipient, policy: .policyDigest.sha256}' \
  /srv/makoto/runs/<run>/reports/verification.json
{
  "decision": "deny",
  "primaryError": "E_ARTIFACT_DIGEST",
  "notPassed": [
    {
      "id": "artifact-bytes",
      "status": "fail"
    }
  ],
  "errors": [
    {
      "step": 12,
      "code": "E_ARTIFACT_DIGEST",
      "message": "final artifact bytes differ from the signed digest"
    }
  ],
  "recipient": "example:analytics-team",
  "policy": "b0b749267a9b3bbfce5dc0a8a9fd570f49525af8a4332c43aa11d9ce1d529a69"
}

Signing that survives retries

Airflow retries tasks. Signed evidence must not fork because of it. sign derives the event identifier from the DAG, run, and statement name, so a retry produces the same event. If the envelope already exists, it is reused only when it covers the same bytes; if the task output changed between attempts, the task fails permanently instead of emitting a second, conflicting history.

def sign(kind: str, role: str, subject: Path, out: Path, *extra: object) -> str:
    """Sign once. A retry reuses the envelope only if it covers the same bytes."""
    if out.exists():
        signed = json.loads(payload(out))["subject"][0]["digest"]["sha256"]
        if signed == sha256(subject):
            return str(out)
        raise AirflowFailException(f"{out.name} already attests different bytes; refusing a second history")
    result = makoto(
        "attest", kind,
        "--subject", f"{subject.name}={subject}",
        *extra,
        "--event-id", event_id(out.name),
        "--key", WORK / "keys" / f"{role}.private.pem",
        "--out", out,
    )
    if result.returncode != 0:
        raise RuntimeError(f"makoto attest {kind} failed: {result.stderr.strip()}")  # retryable
    return str(out)

Both behaviours were exercised against the run above: a retry over identical bytes returned the existing envelope unchanged, and a retry after the raw file changed raised AirflowFailException: 01-origin.dsse.json already attests different bytes; refusing a second history.

What travels through XCom, and what never does

Artifact and envelope paths

Passed through XCom. They are small strings pointing into the run-scoped directory.

Manifest, head, and artifact digests

Returned by create_handoff. In production, publish them through a channel the bundle cannot rewrite and have the receiver read them from there.

Private keys

Never. Tasks read them from the worker filesystem under MAKOTO_WORK/keys; mount them from your secrets backend, one key per role.

Bundles and reports

Never. XCom rows live in the metadata database. The gate writes the report to disk and returns only its path and digests.

The gate, and why its exceptions differ

    def makoto_gate(delivered: str, handoff: dict) -> dict:
        expected_artifact = run_dir("receiver-inbox") / "expected-artifact.json"
        expected_artifact.write_text(json.dumps(handoff["artifact"]))
        result = makoto(
            "verify", "bundle", delivered,
            "--policy", WORK / "receiver" / "policy.json",
            "--expected-manifest", handoff["manifest"],
            "--expected-head", handoff["head"],
            "--expected-artifact", expected_artifact,
            "--expected-recipient", RECIPIENT,
            "--json",
        )
        if result.returncode not in (0, 1):
            raise RuntimeError(f"verifier could not run: {result.stderr.strip()}")  # retryable
        report_path = run_dir("reports") / "verification.json"
        report_path.write_text(result.stdout)
        report = json.loads(result.stdout)
        if report["decision"] != "allow":
            failed = [f"step {e['step']} {e['code']}: {e['message']}" for e in report["errors"]]
            # AirflowFailException skips retries: a denial will not change on retry.
            raise AirflowFailException(
                f"Makoto {report['decision']} ({report['primaryError']}); "
                f"report at {report_path}; " + "; ".join(failed)
            )
        return {
            "decision": report["decision"],
            "manifest": report["manifestDigest"]["sha256"],
            "policy": report["policyDigest"]["sha256"],
            "report": str(report_path),
        }

Two exception types carry two meanings. RuntimeError means the verifier could not run, exit status 2 or 3, which may be transient, so the task’s retries apply. AirflowFailException means the verifier ran and refused the data. Retrying cannot change a denial, and retrying would hide it in the UI, so it fails immediately.

Taking it to production

Split producer and receiver

Run makoto_gate and publish in the receiving team’s own DAG and deployment, with its own policy file. A gate that shares a filesystem with the producer can be satisfied by the producer.

Pin evaluation time for backfills

The gate uses the current time, which enforces the policy’s 24-hour maxAgeSeconds. When re-verifying historical runs, pass --evaluation-time from the run’s data interval and keep the policy snapshot that was in force.

Keep the report

Ship reports/verification.json to durable storage with the policy digest, allowed or not. It is the record of why data was accepted.

Keep the run directory immutable

Never rewrite a signed file in place to append a step. A new step is a new statement that names its predecessor.

What the evidence can say

It can bind one committed task output to exact inputs, predecessor statements, and the key that signed it, and prove that the bytes a receiver holds are the bytes that were signed. It cannot prove the task code executed honestly, that the source system told the truth, or that DAG ownership implies a recipient is authorized. Those remain controls around the pipeline, not properties of the evidence.

Troubleshooting

{"errorClass":"invalid-input","message":"$.rules: array is not in canonical order"}

Policy rules must be sorted by id, compared as bytes. The bootstrap script sorts them before writing.

{"errorClass":"invalid-input","message":"output already exists: ..."} on a retry

The CLI refuses to overwrite an existing envelope unless given --force, and exits 2. The sign helper handles this by reusing a matching envelope; do not add --force to attestation calls, or a retry can silently replace history.

FileNotFoundError: [Errno 2] No such file or directory: 'uv' in a task log

The worker’s PATH does not include uv. Install it in the worker image, or call $MAKOTO_CORE/.venv/bin/makoto directly.

The gate denies with E_HANDOFF_STALE

The handoff is older than the policy’s maxAgeSeconds at evaluation time. Verifying the clean run above two days later with --evaluation-time produces exactly this denial and no other error. Verify promptly after delivery, or pass an explicit --evaluation-time when re-verifying.