Attach provenance where data changes custody or shape.

Four responsibilities

1. ObserveHash the exact input and output artifacts the platform actually read and wrote.
2. AttestCreate an origin or transformation statement with predecessor links and platform metadata.
3. CarryKeep signed statements, required profiles, and final artifact references together at the handoff.
4. VerifyResolve exact schemas, authorize signers, reconstruct the graph, and hash the received bytes.

Common attachment points

The shared boundary helper

Every platform page uses the same two files. makoto_boundary.py is standard-library Python that signs origins and transformations, builds a handoff, and verifies a delivery through the pinned CLI. It turns the three outcomes a pipeline must tell apart into three exceptions: MakotoToolError for a tool that could not run, which may be retried; MakotoConflict for an envelope that already covers different bytes; and MakotoDenied for a verdict, which must never be retried.

"""Makoto boundary helpers for pipeline integrations. Standard library only.

Every function drives the pinned reference CLI, so a platform integration never
reimplements signing, hashing, or verification. Configure with:

    MAKOTO_CORE    reviewed makoto checkout, synced with `uv sync --locked --dev`
    MAKOTO_KEYS    directory holding <role>.private.pem signing keys
    MAKOTO_POLICY  the receiver's trust policy (receiver side only)
"""

from __future__ import annotations

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


class MakotoToolError(RuntimeError):
    """The CLI could not run (exit 2 or 3). Possibly transient; never a verdict."""


class MakotoConflict(RuntimeError):
    """An envelope already exists for different bytes. Retrying cannot fix this."""


class MakotoDenied(RuntimeError):
    """The verifier ran and refused the data. Carries the full report."""

    def __init__(self, report: dict, report_path: Path) -> None:
        errors = "; ".join(f"step {e['step']} {e['code']}: {e['message']}" for e in report["errors"])
        super().__init__(f"Makoto {report['decision']} ({report['primaryError']}): {errors}")
        self.report = report
        self.report_path = report_path


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


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with Path(path).open("rb") as stream:
        for block in iter(lambda: stream.read(1 << 20), b""):
            digest.update(block)
    return digest.hexdigest()


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


def statement_digest(envelope: Path) -> str:
    """Statement identity is the SHA-256 of the decoded payload bytes."""
    return hashlib.sha256(payload(envelope)).hexdigest()


def _sign(kind: str, role: str, subject: Path, out: Path, event_id: str, *extra: object) -> Path:
    subject, out = Path(subject).resolve(), Path(out).resolve()
    if out.exists():
        signed = json.loads(payload(out))["subject"][0]["digest"]["sha256"]
        if signed == sha256_file(subject):
            return out  # a retry after a successful signature
        raise MakotoConflict(f"{out.name} already attests different bytes")
    out.parent.mkdir(parents=True, exist_ok=True)
    result = makoto(
        "attest", kind,
        "--subject", f"{subject.name}={subject}",
        *extra,
        "--event-id", event_id,
        "--key", Path(os.environ["MAKOTO_KEYS"]) / f"{role}.private.pem",
        "--out", out,
    )
    if result.returncode != 0:
        raise MakotoToolError(f"makoto attest {kind}: {result.stderr.strip()}")
    return out


def sign_origin(subject: Path, out: Path, *, source_kind: str, source_uri: str, event_id: str,
                source_metadata: dict | None = None, role: str = "origin") -> Path:
    """source_metadata may carry name, mediaType, retrievedAt, and version."""
    extra: list[object] = ["--source-kind", source_kind, "--source-uri", source_uri]
    if source_metadata:
        metadata = Path(out).resolve().parent / f"{Path(out).name}.source.json"
        metadata.parent.mkdir(parents=True, exist_ok=True)
        metadata.write_text(json.dumps(source_metadata))
        extra += ["--source-metadata", metadata]
    return _sign("origin", role, subject, out, event_id, *extra)


def sign_transform(subject: Path, out: Path, *, inputs: list[tuple[Path, Path]], operation_type: str,
                   event_id: str, operation_name: str | None = None, role: str = "transform") -> Path:
    """inputs: (input artifact path, envelope that attests that exact artifact)."""
    out = Path(out).resolve()
    extra: list[object] = ["--operation-type", operation_type]
    if operation_name:
        extra += ["--operation-name", operation_name]
    for index, (artifact, predecessor) in enumerate(inputs):
        artifact, predecessor = Path(artifact).resolve(), Path(predecessor).resolve()
        binding = out.parent / f"{out.name}.input-{index}.json"
        binding.parent.mkdir(parents=True, exist_ok=True)
        binding.write_text(json.dumps({
            "name": f"input-{index}",
            "path": str(artifact),
            "predecessor": str(predecessor),
            "subjectName": artifact.name,
        }))
        extra += ["--input-binding", binding]
    return _sign("transform", role, subject, out, event_id, *extra)


def create_handoff(head: Path, final: Path, attestations: Path, out: Path, *, recipient: str,
                   bundle_id: str, media_type: str, role: str = "handoff") -> dict:
    """Write a bundle and return the anchors the receiver must obtain out of band."""
    head, final, out = Path(head).resolve(), Path(final).resolve(), Path(out).resolve()
    binding = out.parent / f"{out.name}.final-artifact.json"
    binding.write_text(json.dumps({
        "head": str(head), "subjectName": final.name, "path": str(final), "mediaType": media_type,
    }))
    result = makoto(
        "handoff", "create",
        "--head", head,
        "--attestations", Path(attestations).resolve(),
        "--artifact-binding", binding,
        "--recipient", recipient,
        "--bundle-id", bundle_id,
        "--key", Path(os.environ["MAKOTO_KEYS"]) / f"{role}.private.pem",
        "--out", out,
        "--force",  # rebuilding from the same signed inputs is safe
    )
    if result.returncode != 0:
        raise MakotoToolError(f"makoto handoff create: {result.stderr.strip()}")
    manifest = json.loads((out / "bundle.json").read_text())["manifest"]
    return {
        "manifest": "sha256:" + statement_digest(out / manifest),
        "head": "sha256:" + statement_digest(head),
        "artifact": {
            "head": {"sha256": statement_digest(head)},
            "subjectName": final.name,
            "digest": {"sha256": sha256_file(final)},
        },
    }


def verify(bundle: Path, anchors: dict, *, recipient: str, report_path: Path,
           evaluation_time: str | None = None) -> dict:
    """Return the report on allow; raise MakotoDenied or MakotoToolError otherwise."""
    report_path = Path(report_path).resolve()
    report_path.parent.mkdir(parents=True, exist_ok=True)
    expected_artifact = report_path.with_name(report_path.stem + ".expected-artifact.json")
    expected_artifact.write_text(json.dumps(anchors["artifact"]))
    args: list[object] = [
        "verify", "bundle", Path(bundle).resolve(),
        "--policy", os.environ["MAKOTO_POLICY"],
        "--expected-manifest", anchors["manifest"],
        "--expected-head", anchors["head"],
        "--expected-artifact", expected_artifact,
        "--expected-recipient", recipient,
        "--json",
    ]
    if evaluation_time:
        args += ["--evaluation-time", evaluation_time]
    result = makoto(*args)
    if result.returncode not in (0, 1):
        raise MakotoToolError(f"makoto verify: {result.stderr.strip()}")
    report_path.write_text(result.stdout)
    report = json.loads(result.stdout)
    if (result.returncode == 0) != (report["decision"] == "allow"):
        raise MakotoToolError("exit status and report decision disagree")
    if report["decision"] != "allow":
        raise MakotoDenied(report, report_path)
    return report

makoto_bootstrap.py creates one Ed25519 key per claim role and a receiver trust policy that authorizes each key for exactly one kind of claim, then runs makoto policy check on the result.

"""Create per-role signing keys and a receiver trust policy for one pipeline.

    MAKOTO_CORE=/opt/makoto uv run makoto_bootstrap.py /srv/makoto \
        --source-kind urn:example:source:orders-export \
        --source-uri urn:example:orders:daily \
        --operation-type urn:example:operation:normalize-orders
"""

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

ROLES = ("origin", "transform", "handoff")
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,
}


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


parser = argparse.ArgumentParser()
parser.add_argument("home", type=Path)
parser.add_argument("--source-kind", required=True)
parser.add_argument("--source-uri", required=True)
parser.add_argument("--operation-type", action="append", required=True)
args = parser.parse_args()

keys_dir = (args.home / "keys").resolve()
keys_dir.mkdir(parents=True, exist_ok=True)
keys, keyid = {}, {}
for role in ROLES:
    private, public = keys_dir / f"{role}.private.pem", keys_dir / f"{role}.public.pem"
    if not private.exists():
        makoto("key", "generate", "--private-out", private, "--public-out", public)
    keyid[role] = json.loads(makoto("key", "inspect", "--public", public, "--json"))["keyid"]
    body = "".join(public.read_text().strip().splitlines()[1:-1])  # base64 SPKI DER
    keys[keyid[role]] = {"type": "ed25519", "publicKey": base64.b64encode(base64.b64decode(body)).decode(),
                         "label": f"{role} signing key"}

rules = [
    {
        "id": "urn:makoto:rule:origin",
        "predicateTypes": ["https://usemakoto.dev/predicate/v0.2/origin"],
        "authorizedKeyIds": [keyid["origin"]],
        "minimumSignatures": 1,
        "sourceKinds": [args.source_kind],
        "sourceUris": [args.source_uri],
    },
    {
        "id": "urn:makoto:rule:transform",
        "predicateTypes": ["https://usemakoto.dev/predicate/v0.2/transform"],
        "authorizedKeyIds": [keyid["transform"]],
        "minimumSignatures": 1,
        "operationTypes": sorted(set(args.operation_type), key=str.encode),
    },
]
policy = {
    "version": "0.2",
    "keys": keys,
    "rules": sorted(rules, key=lambda rule: rule["id"].encode()),  # canonical order is required
    "handoff": {
        "authorizedKeyIds": [keyid["handoff"]],
        "minimumSignatures": 1,
        "requireExpectedManifest": True,
        "requireExpectedHead": True,
        "requireExpectedArtifacts": True,
        "requireRecipient": True,
        "requireNonce": False,
        "allowReplayableHandoff": False,
        "maxAgeSeconds": 86400,
        "maxFutureSkewSeconds": 300,
    },
    "requiredProfiles": [],
    "limits": LIMITS,
}
policy_path = (args.home / "policy.json").resolve()
policy_path.write_text(json.dumps(policy, indent=2) + "\n")
makoto("policy", "check", "--policy", policy_path)
print(f"export MAKOTO_KEYS={keys_dir}")
print(f"export MAKOTO_POLICY={policy_path}")

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
uv run makoto_bootstrap.py /srv/makoto \
  --source-kind urn:example:source:orders-export \
  --source-uri urn:example:orders:daily \
  --operation-type urn:example:operation:normalize-orders
export MAKOTO_KEYS=/srv/makoto/keys MAKOTO_POLICY=/srv/makoto/policy.json

Exercised directly, the helper allows a clean handoff, refuses a changed file, refuses the wrong recipient, and refuses to re-sign changed bytes:

allow
Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest
Makoto deny (E_HANDOFF_RECIPIENT): step 11 E_HANDOFF_RECIPIENT: handoff recipient does not match
conflict: 01-origin.dsse.json already attests different bytes

How each pattern was tested

Airflow

DAG run with airflow dags test on Airflow 3.3.1, clean and tampered; retry reuse and conflict exercised.

Dagster

Assets materialized with Dagster 1.13.22; the blocking check stopped downstream assets on a denial.

Prefect

Flow run on Prefect 3.8.6; the gate failed without retrying.

dbt

Built with dbt-core 1.12.5 and dbt-duckdb 1.11.0 through dbtRunner.

Spark

Submitted to PySpark 4.2.0 in local mode on Java 21.

Kafka

Run against an Apache Kafka 4.1.0 broker in a container: one allowed window, two refusals.

Snowflake

Makoto path executed with a stubbed connector; not run against an account.

Databricks

Task scripts executed on local PySpark with a stubbed runtime; not deployed to a workspace.

Expanso

Pipeline passed expanso-edge validate; seal and collector executed; not deployed to an edge node.

Platform integrations

Each page starts from a real extension surface in the platform and shows the code that puts Makoto there, what a run prints, and what a gate failure looks like. A maintained adapter would still need an owner, tests in the platform’s own CI, and a release.

1. 01

   Apache Airflow

   A TaskFlow DAG that signs each task boundary and fails the run with AirflowFailException when the gate denies the handoff.

   Read the integration →
2. 02

   dbt

   A dbtRunner wrapper that exports a model to a file, signs it against its seed, and refuses the handoff before loading.

   Read the integration →
3. 03

   Apache Spark

   A driver that waits for _SUCCESS, signs the committed part file, and exits non-zero on a denial.

   Read the integration →
4. 04

   Apache Kafka

   Close an offset window, sign its bytes, and commit consumer offsets only after the verifier and a window check both pass.

   Read the integration →
5. 05

   Snowflake

   Unload a pinned snapshot, sign the downloaded file, and refuse the receiving load before PUT.

   Read the integration →
6. 06

   Databricks

   A three-task job: export a pinned Delta version and sign it, verify, and publish only after the gate succeeds.

   Read the integration →
7. 07

   Dagster

   Assets sign what they materialize; a blocking asset check stops downstream assets on a denial.

   Read the integration →
8. 08

   Prefect

   Tasks sign persisted files; the gate task fails the flow and a retry condition refuses to retry a verdict.

   Read the integration →
9. 09

   Expanso

   An Expanso Edge command processor seals each batch at the edge; a collector admits batches only on an allow.

   Read the integration →