Gate a Python ingestion job on a Makoto verdict.

1. Install the verifier and the fixtures

Clone the core repository at the documented revision, sync its locked environment, and generate the fixture set every example uses.

git clone https://github.com/makoto-project/makoto.git
cd makoto
git checkout 276008eeec0f10d5322c4c627b8c103222fee841
uv sync --locked --dev

# Writes the deterministic fixture set: one allowed bundle, seven denials.
./scripts/demo.sh --acceptance

export MAKOTO_CORE="$PWD"

The program declares its own dependency, jsonschema, in inline script metadata, so uv run provisions it without a project file. With plain pip, install jsonschema>=4.25 first.

2. The complete program

Save this as makoto_gate.py anywhere outside the core checkout. The sections that follow walk through each part.

# /// script
# requires-python = ">=3.11"
# dependencies = ["jsonschema>=4.25,<5"]
# ///
"""Validate, attest, verify, and gate on a Makoto report from Python."""

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path

from jsonschema import Draft202012Validator

CORE = Path(os.environ["MAKOTO_CORE"]).resolve()
DEMO = CORE / "demos/end-to-end/generated"
SCHEMA_BASE = "https://usemakoto.dev/schema/v0.2/"

# Independent expectations. In production these arrive out of band from the
# producer (a ticket, a signed release note, a registry), never from the bundle.
EXPECTED_MANIFEST = "sha256:e24aae77ba1374162edff49aa0ab1d8dd7e38c333e134a2128ad1598fec99fb7"
EXPECTED_HEAD = "sha256:99aee76db46a720405da6d0015d427c791b0d4e02367b9e2bc36ce82ae812bce"
EVALUATION_TIME = "2026-09-16T16:00:00Z"


class MakotoToolError(RuntimeError):
    """Exit 2 or 3: the verifier could not run. This is never a verdict."""


class MakotoDenied(RuntimeError):
    def __init__(self, report: dict) -> None:
        super().__init__(f"{report['decision']}: {report['primaryError']}")
        self.report = report


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


def validate_structure(instance_path: Path, schema_name: str) -> list[str]:
    # The schema host rejects urllib's default User-Agent with 403.
    request = urllib.request.Request(SCHEMA_BASE + schema_name, headers={"User-Agent": "makoto-gate"})
    with urllib.request.urlopen(request, timeout=10) as response:
        schema = json.load(response)
    instance = json.loads(instance_path.read_text(encoding="utf-8"))
    validator = Draft202012Validator(schema)
    return [
        f"{'/'.join(map(str, error.absolute_path)) or '(root)'}: {error.message}"
        for error in validator.iter_errors(instance)
    ]


def attest_origin(artifact: Path, work: Path) -> Path:
    private_key, public_key = work / "producer.private.pem", work / "producer.public.pem"
    out = work / "origin.dsse.json"
    for args in (
        ("key", "generate", "--private-out", str(private_key), "--public-out", str(public_key)),
        (
            "attest", "origin",
            "--subject", f"{artifact.name}={artifact}",
            "--source-kind", "urn:example:source:nightly-export",
            "--source-uri", "urn:example:orders:2026-09-16",
            "--key", str(private_key),
            "--out", str(out),
        ),
    ):
        completed = makoto(*args)
        if completed.returncode != 0:
            raise MakotoToolError(completed.stderr.strip())
    return out


def verify_bundle(bundle: Path) -> dict:
    completed = makoto(
        "verify", "bundle", str(bundle),
        "--policy", str(DEMO / "receiver/policy.json"),
        "--schema-catalog", str(DEMO / "receiver/catalog.json"),
        "--expected-manifest", EXPECTED_MANIFEST,
        "--expected-head", EXPECTED_HEAD,
        "--expected-artifact", str(DEMO / "receiver/expected-artifact.json"),
        "--evaluation-time", EVALUATION_TIME,
        "--json",
    )
    if completed.returncode not in (0, 1):
        # stderr carries one JSON object: {"errorClass": ..., "message": ...}
        raise MakotoToolError(completed.stderr.strip())
    try:
        report = json.loads(completed.stdout)
    except json.JSONDecodeError as error:
        raise MakotoToolError(f"verifier emitted a malformed report: {error}") from error
    # Exit status and decision must agree. If they do not, trust neither.
    if (completed.returncode == 0) != (report.get("decision") == "allow"):
        raise MakotoToolError("exit status and report decision disagree")
    if report["decision"] != "allow":
        raise MakotoDenied(report)
    return report


def explain(report: dict) -> None:
    print(f"decision={report['decision']} primaryError={report['primaryError']}")
    for check in report["checks"]:
        if check["status"] != "pass":
            print(f"  check {check['id']}: {check['status']} {check['prerequisiteChecks']}")
    for error in report["errors"]:
        print(f"  step {error['step']} {error['code']}: {error['message']}")
    summary = report["summary"]
    print(
        f"  statements {summary['statementsValid']}/{summary['statementsTotal']} valid, "
        f"signatures {summary['signaturesValid']}/{summary['signaturesTotal']} valid, "
        f"policy {report['policyDigest']['sha256'][:12]}"
    )


def main() -> int:
    problems = validate_structure(DEMO / "receiver/policy.json", "trust-policy.schema.json")
    print("policy structure:", "valid" if not problems else problems)

    with tempfile.TemporaryDirectory() as scratch:
        work = Path(scratch)
        artifact = work / "orders.csv"
        artifact.write_text("order_id,total\n1001,42.00\n", encoding="utf-8")
        envelope = attest_origin(artifact, work)
        print("signed", json.loads(makoto("envelope", "inspect", "--envelope", str(envelope), "--json").stdout))

        report = verify_bundle(DEMO / "positive-bundle")
        explain(report)

        tampered = work / "tampered-bundle"
        shutil.copytree(DEMO / "positive-bundle", tampered)
        final = next((tampered / "artifacts/final").glob("*.bin"))
        final.write_bytes(final.read_bytes().replace(b"]", b",{}]", 1))
        try:
            verify_bundle(tampered)
        except MakotoDenied as denied:
            explain(denied.report)
            if denied.report["primaryError"] != "E_ARTIFACT_DIGEST":
                return 1
            print("tampered bundle refused, as it should be")
            return 0
    print("tampered bundle was accepted", file=sys.stderr)
    return 1


if __name__ == "__main__":
    sys.exit(main())

MAKOTO_CORE=/path/to/makoto uv run makoto_gate.py

One run’s output, unedited. Key IDs, payload digests, and temporary paths differ on every run; the decisions, codes, and policy digest do not.

policy structure: valid
signed {'payloadDigest': {'sha256': '0411bf08b3fb75a8c723d3b8a4e93ae03e3150a163930785aeb9eb6ddbec77aa'}, 'payloadType': 'application/vnd.in-toto+json', 'signatureKeyIds': ['sha256:1090f3864e8d67bde4dcdf76e03adfc1815e6b2a59f7b2392afac448c31a64e2']}
decision=allow primaryError=None
  statements 3/3 valid, signatures 4/4 valid, policy 614eb137303b
decision=deny primaryError=E_ARTIFACT_DIGEST
  check artifact-bytes: fail []
  step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest
  statements 3/3 valid, signatures 4/4 valid, policy 614eb137303b
tampered bundle refused, as it should be

3. Validate structure against the hosted schema

Structural validation is cheap, keyless, and belongs at the ingestion boundary, where it rejects a malformed policy or statement before anything is spent on cryptography. It answers whether a document is shaped correctly and nothing else: a schema-valid policy can still trust the wrong key.

def validate_structure(instance_path: Path, schema_name: str) -> list[str]:
    # The schema host rejects urllib's default User-Agent with 403.
    request = urllib.request.Request(SCHEMA_BASE + schema_name, headers={"User-Agent": "makoto-gate"})
    with urllib.request.urlopen(request, timeout=10) as response:
        schema = json.load(response)
    instance = json.loads(instance_path.read_text(encoding="utf-8"))
    validator = Draft202012Validator(schema)
    return [
        f"{'/'.join(map(str, error.absolute_path)) or '(root)'}: {error.message}"
        for error in validator.iter_errors(instance)
    ]

The same call works for any hosted schema: pass envelope.schema.json for a DSSE envelope or verification-report.schema.json for a stored report. The validator catalogue covers the other tools.

4. Generate an attestation

A producer signs an origin statement where data enters its system. key generate writes an Ed25519 key pair with the private half at mode 0600; attest origin hashes the exact subject bytes and writes a DSSE envelope. Keep the private key inside the job’s secret boundary.

def attest_origin(artifact: Path, work: Path) -> Path:
    private_key, public_key = work / "producer.private.pem", work / "producer.public.pem"
    out = work / "origin.dsse.json"
    for args in (
        ("key", "generate", "--private-out", str(private_key), "--public-out", str(public_key)),
        (
            "attest", "origin",
            "--subject", f"{artifact.name}={artifact}",
            "--source-kind", "urn:example:source:nightly-export",
            "--source-uri", "urn:example:orders:2026-09-16",
            "--key", str(private_key),
            "--out", str(out),
        ),
    ):
        completed = makoto(*args)
        if completed.returncode != 0:
            raise MakotoToolError(completed.stderr.strip())
    return out

A transformation step uses attest transform with one --input-binding file per input, naming the predecessor envelope and the exact subject it consumed. See the transform predicate for the binding fields.

5. Verify a bundle and handle a denial

Where the expected digests come from. The manifest digest, head digest, and expected-artifact file are the receiver’s independent anchors. The fixture policy sets requireExpectedManifest, requireExpectedHead, and requireExpectedArtifacts, so leaving them out is a denial, not a shortcut. In production they arrive through a channel the producer’s bundle cannot rewrite: a change ticket, a signed release note, a registry entry. The values below are the fixture’s published digests.

class MakotoToolError(RuntimeError):
    """Exit 2 or 3: the verifier could not run. This is never a verdict."""


class MakotoDenied(RuntimeError):
    def __init__(self, report: dict) -> None:
        super().__init__(f"{report['decision']}: {report['primaryError']}")
        self.report = report

def verify_bundle(bundle: Path) -> dict:
    completed = makoto(
        "verify", "bundle", str(bundle),
        "--policy", str(DEMO / "receiver/policy.json"),
        "--schema-catalog", str(DEMO / "receiver/catalog.json"),
        "--expected-manifest", EXPECTED_MANIFEST,
        "--expected-head", EXPECTED_HEAD,
        "--expected-artifact", str(DEMO / "receiver/expected-artifact.json"),
        "--evaluation-time", EVALUATION_TIME,
        "--json",
    )
    if completed.returncode not in (0, 1):
        # stderr carries one JSON object: {"errorClass": ..., "message": ...}
        raise MakotoToolError(completed.stderr.strip())
    try:
        report = json.loads(completed.stdout)
    except json.JSONDecodeError as error:
        raise MakotoToolError(f"verifier emitted a malformed report: {error}") from error
    # Exit status and decision must agree. If they do not, trust neither.
    if (completed.returncode == 0) != (report.get("decision") == "allow"):
        raise MakotoToolError("exit status and report decision disagree")
    if report["decision"] != "allow":
        raise MakotoDenied(report)
    return report

Three rules keep this fail-closed. Exit status 2 or 3 raises MakotoToolError, because no verdict exists. A report that does not parse is a tool failure, not a denial. And the exit status must agree with decision; if they ever disagree, trust neither.

The caller decides what a denial means for the job. Here the program copies the good bundle, appends bytes to the delivered artifact, and expects the verifier to refuse it with E_ARTIFACT_DIGEST:

        tampered = work / "tampered-bundle"
        shutil.copytree(DEMO / "positive-bundle", tampered)
        final = next((tampered / "artifacts/final").glob("*.bin"))
        final.write_bytes(final.read_bytes().replace(b"]", b",{}]", 1))
        try:
            verify_bundle(tampered)
        except MakotoDenied as denied:
            explain(denied.report)
            if denied.report["primaryError"] != "E_ARTIFACT_DIGEST":
                return 1
            print("tampered bundle refused, as it should be")
            return 0

6. Read the verification report

Read decision first, then primaryError for the boundary that failed, then checks for what was actually established. errors carries the step number and a typed context. Store policyDigest with the report: a verdict is only meaningful under the exact policy that produced it. The full shape is the verification-report schema.

def explain(report: dict) -> None:
    print(f"decision={report['decision']} primaryError={report['primaryError']}")
    for check in report["checks"]:
        if check["status"] != "pass":
            print(f"  check {check['id']}: {check['status']} {check['prerequisiteChecks']}")
    for error in report["errors"]:
        print(f"  step {error['step']} {error['code']}: {error['message']}")
    summary = report["summary"]
    print(
        f"  statements {summary['statementsValid']}/{summary['statementsTotal']} valid, "
        f"signatures {summary['signaturesValid']}/{summary['signaturesTotal']} valid, "
        f"policy {report['policyDigest']['sha256'][:12]}"
    )

The tampered run reports one failing check, artifact-bytes, at step 12. Signatures, authorization, and the graph all still pass, which is the point: the claims are intact and the data under them changed. The ordered checks list every code by step.

In-process: the reference library

The command-line boundary is the stable contract. If a long-running Python service cannot afford a subprocess per bundle, it can call the same verifier in-process from inside the synced checkout. VerificationRequest and verify_bundle are the functions the CLI itself calls, and they return the identical report dictionary.

"""Run the reference verifier in-process. Run from inside the core checkout:
uv run python in_process.py
"""

from pathlib import Path

import makoto
from makoto.bundle import VerificationRequest, verify_bundle

core = Path(makoto.__file__).resolve().parents[2]
demo = core / "demos/end-to-end/generated"
expected_artifact = {
    "head": {"sha256": "99aee76db46a720405da6d0015d427c791b0d4e02367b9e2bc36ce82ae812bce"},
    "subjectName": "customers.public.json",
    "digest": {"sha256": "ecd821dbe6f4204c4aad64878643d7954cb8c94abbf8d0b28f9c2bfc272f781e"},
}

report = verify_bundle(
    VerificationRequest(
        bundle_root=demo / "positive-bundle",
        policy_path=demo / "receiver/policy.json",
        repository_root=core,
        schema_catalogs=(demo / "receiver/catalog.json",),
        expected_manifest={"sha256": "e24aae77ba1374162edff49aa0ab1d8dd7e38c333e134a2128ad1598fec99fb7"},
        expected_heads=({"sha256": "99aee76db46a720405da6d0015d427c791b0d4e02367b9e2bc36ce82ae812bce"},),
        expected_artifacts=(expected_artifact,),
        evaluation_time="2026-09-16T16:00:00Z",
    )
)
print(report["decision"], report["primaryError"])

cd "$MAKOTO_CORE" && uv run python /path/to/in_process.py
# allow None

Pin before you import. makoto.bundle is an internal module of the reference implementation, not a published API. Its fields can change between revisions, so lock the core checkout to a reviewed commit and rerun this fixture when you move it. The CLI path above does not carry that coupling.

Exit status contract

0 · allow

Every required check passed under this exact policy and expectation set. stdout holds the report; keep it with the policy digest.

1 · deny or indeterminate

stdout still holds a complete report. Refuse the data and route on primaryError.

2 · invalid input

A missing file, a malformed flag, an unreadable policy. No report is written; stderr carries one JSON object with errorClass and message.

3 · internal error

Same stderr shape with errorClass set to internal. Treat it as an outage, never as a verdict.

Troubleshooting

urllib.error.HTTPError: HTTP Error 403: Forbidden when fetching a schema

The schema host rejects the default Python-urllib User-Agent. Send any explicit User-Agent, as validate_structure does.

Every run denies with E_EXPECTED_MANIFEST, E_EXPECTED_HEAD, and E_EXPECTED_ARTIFACT at step 11

The expectation flags are missing or stale. The freshness-anchors check fails because the policy requires all three. Supply the digests the producer published for this handoff.

Exit status 2 with {"errorClass":"invalid-input",...} on stderr

An input path is wrong or unreadable, or a singleton flag was passed twice. Paths are resolved from the process working directory, so pass absolute paths when the verifier runs with cwd set to the core checkout.

The report shows skipped for a check you care about

An earlier check failed and did not establish its prerequisite. prerequisiteChecks names it. A skipped check was not tested; never read it as a pass.

uv: command not found inside a service or CI job

The binding needs uv on PATH for the process that spawns the verifier. Install it in the image or call the checkout’s .venv/bin/makoto directly after uv sync --locked.