Dagster: materialization is the hook; exact artifacts are the evidence.

Where Makoto sits

Signing happens inside the asset that committed the bytes, so the materialization and its evidence share one run ID. The gate is an asset check, which Dagster already treats as a first-class verdict on an asset.

1. orders_rawCommits the raw export and signs an origin statement. The statement digest is recorded as materialization metadata.
2. orders_normalizedTransforms, signs a transformation statement bound to the raw bytes and origin statement, and builds the handoff bundle.
3. makoto_handoff_verifiesA blocking=True asset check. It verifies the delivered bundle and returns a failed AssetCheckResult with ERROR severity on a denial.
4. orders_publishedDepends on orders_normalized, so a failed blocking check prevents it from materializing.

1. Install

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. Put both next to the code below.

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 pip install "dagster==1.13.22"

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

2. The assets and the check

Save this as orders_assets.py.

"""Dagster assets that sign what they materialize and a blocking Makoto check."""

import json
import os
import shutil
import uuid
from pathlib import Path

import dagster as dg

import makoto_boundary as mb

WORK = Path(os.environ["MAKOTO_WORK"])
RECIPIENT = "example:analytics-team"


def run_dir(context, *parts: str) -> Path:
    path = WORK.joinpath("runs", context.run_id, *parts)
    path.mkdir(parents=True, exist_ok=True)
    return path


def event_id(context, name: str) -> str:
    return f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{context.run_id}/{name}')}"


@dg.asset
def orders_raw(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    raw = run_dir(context, "data") / "orders.raw.csv"
    raw.write_text("order_id,region,total\n1001, us ,42.00\n1002,GB,17.50\n")
    envelope = mb.sign_origin(
        raw, run_dir(context, "attestations") / "01-origin.dsse.json",
        source_kind="urn:example:source:orders-export",
        source_uri="urn:example:orders:daily",
        event_id=event_id(context, "origin"),
    )
    return dg.MaterializeResult(metadata={
        "path": dg.MetadataValue.path(str(raw)),
        "sha256": mb.sha256_file(raw),
        "makoto_statement": mb.statement_digest(envelope),
    })


@dg.asset(deps=[orders_raw])
def orders_normalized(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    raw = run_dir(context, "data") / "orders.raw.csv"
    header, *rows = raw.read_text().splitlines()
    out = run_dir(context, "data") / "orders.normalized.csv"
    out.write_text("\n".join([header, *(",".join(c.strip().lower() for c in r.split(",")) for r in rows)]) + "\n")
    attestations = run_dir(context, "attestations")
    head = mb.sign_transform(
        out, attestations / "02-normalize.dsse.json",
        inputs=[(raw, attestations / "01-origin.dsse.json")],
        operation_type="urn:example:operation:normalize-orders",
        event_id=event_id(context, "normalize"),
    )
    anchors = mb.create_handoff(
        head, out, attestations, run_dir(context) / "bundle",
        recipient=RECIPIENT, bundle_id=event_id(context, "bundle"), media_type="text/csv",
    )
    # Stand-in for an out-of-band channel the bundle cannot rewrite.
    (run_dir(context) / "anchors.json").write_text(json.dumps(anchors))
    return dg.MaterializeResult(metadata={
        "path": dg.MetadataValue.path(str(out)),
        "makoto_statement": mb.statement_digest(head),
        "makoto_manifest": anchors["manifest"],
    })


@dg.asset_check(asset=orders_normalized, blocking=True)
def makoto_handoff_verifies(context: dg.AssetCheckExecutionContext) -> dg.AssetCheckResult:
    op = context.op_execution_context  # carries run_id; the check context does not
    delivered = run_dir(op, "receiver") / "bundle"
    shutil.copytree(run_dir(op) / "bundle", delivered, dirs_exist_ok=True)
    if os.environ.get("MAKOTO_TAMPER_IN_TRANSIT") == "1":
        final = next((delivered / "artifacts" / "final").glob("*.bin"))
        final.write_bytes(final.read_bytes() + b"1003,us,0.01\n")
    anchors = json.loads((run_dir(op) / "anchors.json").read_text())
    report_path = run_dir(op, "receiver") / "verification.json"
    try:
        report = mb.verify(delivered, anchors, recipient=RECIPIENT, report_path=report_path)
    except mb.MakotoDenied as denied:
        return dg.AssetCheckResult(
            passed=False,
            severity=dg.AssetCheckSeverity.ERROR,
            description=str(denied),
            metadata={
                "decision": denied.report["decision"],
                "primary_error": denied.report["primaryError"],
                "report": dg.MetadataValue.path(str(report_path)),
            },
        )
    return dg.AssetCheckResult(passed=True, metadata={
        "decision": report["decision"],
        "policy_sha256": report["policyDigest"]["sha256"],
        "report": dg.MetadataValue.path(str(report_path)),
    })


@dg.asset(deps=[orders_normalized])
def orders_published(context: dg.AssetExecutionContext) -> None:
    context.log.info("orders_normalized passed the Makoto gate; publishing")


defs = dg.Definitions(
    assets=[orders_raw, orders_normalized, orders_published],
    asset_checks=[makoto_handoff_verifies],
)

3. Run it

export MAKOTO_WORK=/srv/makoto/work
dagster asset materialize -m orders_assets --select '*'

Event log, run and step IDs trimmed:

orders_raw - STEP_SUCCESS - step finished
orders_normalized - STEP_SUCCESS - step finished
orders_normalized_makoto_handoff_verifies - ASSET_CHECK_EVALUATION - Asset check 'makoto_handoff_verifies' on 'orders_normalized' passed.
orders_normalized_makoto_handoff_verifies - STEP_SUCCESS - step finished
orders_published - orders_normalized passed the Makoto gate; publishing
orders_published - STEP_SUCCESS - step finished
RUN_SUCCESS - Finished execution of run for "__ASSET_JOB".

4. What a gate failure looks like

MAKOTO_TAMPER_IN_TRANSIT=1 dagster asset materialize -m orders_assets --select '*'

orders_raw - STEP_SUCCESS - step finished
orders_normalized - STEP_SUCCESS - step finished
orders_normalized_makoto_handoff_verifies - ASSET_CHECK_EVALUATION - Asset check 'makoto_handoff_verifies' on 'orders_normalized' did not pass. Description: 'Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest'
orders_normalized_makoto_handoff_verifies - STEP_FAILURE - Execution of step "orders_normalized_makoto_handoff_verifies" failed.
RUN_FAILURE - Execution of run for "__ASSET_JOB" failed. Steps failed: ['orders_normalized_makoto_handoff_verifies'].
dagster._core.errors.DagsterAssetCheckFailedError: 1 blocking asset check failed with ERROR severity:

The check fails, the run fails, and orders_published never executes. The check’s metadata carries primary_error and a path to the full report, so the asset page in the Dagster UI shows why.

Design notes

Why a check, not an exception in the asset

An asset check records a verdict against the asset itself and keeps its history. Raising inside orders_normalized would mark the transformation as broken when the transformation succeeded and the delivery did not.

The check context has no run ID

AssetCheckExecutionContext does not expose run_id. Use context.op_execution_context, as the check does, to reach the same run directory as the asset.

Partitions

Sign one artifact per partition key and include the key in the event ID, so a backfill of one partition cannot collide with another.

Other attachment points on this platform

Asset materialization

After storage commits, resolve exact output bytes or a deterministic dataset manifest and append a transformation statement.

Asset observation

Use an observation as source context only when the integration can bind it to exact observed bytes.

Partitions

Declare one bounded artifact identity per partition or partition set; avoid treating an open-ended logical asset as immutable.

Metadata

Carry run ID, asset key, partition key, code version, and checks as operation metadata without assuming they authenticate the data.

What the evidence can say

It can bind one committed materialization to exact inputs, output, and selected Dagster context. It cannot prove the asset function executed as claimed or convert every logical asset dependency into a complete signed history automatically.

Troubleshooting

AttributeError: 'AssetCheckExecutionContext' object has no attribute 'run_id'

Reach run-scoped values through context.op_execution_context.

Downstream assets still materialize after a failed check

The check must be declared with blocking=True and must return severity=AssetCheckSeverity.ERROR. With WARN, the same tampered run above ended in RUN_SUCCESS and orders_published materialized.

makoto_boundary.MakotoConflict on a re-execution

The run directory already holds an envelope for different bytes. Re-execute from a new run, or investigate why the asset produced different output for the same run.