Kafka: close a window before you sign it.

Where Makoto sits

The signed artifact is a deterministic rendering of the window: message values in offset order, one per line. Producer and consumer build it with the same function, so the consumer can hash what it actually read.

1. Produce and flushIdempotent producer; the window is closed only when every delivery report has returned.
2. Sign the windowAn origin statement over the window bytes, with partition=0;offsets=0-2 as source metadata.
3. AnnounceA notice carries the window, the bundle location, and the anchors. In production it travels through a channel the receiver trusts independently.
4. Consume the windowAuto-commit is off. The consumer reads exactly the announced offsets and hashes what it got.
5. Verify, then check the windowThe verifier compares the consumer’s digest with the signed artifact; the consumer then checks the signed offsets against the offsets it read. Only then are offsets committed.

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 "confluent-kafka==2.15.1"

uv run makoto_bootstrap.py /srv/makoto \
  --source-kind urn:example:source:kafka-offset-window \
  --source-uri kafka://localhost:9092/orders \
  --operation-type urn:example:operation:none
export MAKOTO_KEYS=/srv/makoto/keys MAKOTO_POLICY=/srv/makoto/policy.json

2. The producer

Save this as window_producer.py.

"""Produce a batch, close its offset window, sign the window, and announce it.

    python window_producer.py /srv/makoto/windows
"""

import json
import sys
import uuid
from pathlib import Path

from confluent_kafka import Producer

import makoto_boundary as mb

BROKER = "localhost:9092"
TOPIC, PARTITION = "orders", 0
RECIPIENT = "example:analytics-team"


def window_bytes(values: list[bytes]) -> bytes:
    """The window artifact: message values in offset order, one per line.
    Producer and consumer must build it identically."""
    return b"".join(value + b"\n" for value in values)


def produce(root: Path) -> None:
    producer = Producer({"bootstrap.servers": BROKER, "enable.idempotence": True})
    delivered: list[tuple[int, bytes]] = []

    def on_delivery(err, msg):
        if err is not None:
            raise RuntimeError(err)
        delivered.append((msg.offset(), msg.value()))

    for order in ({"order_id": 1001, "region": "us", "total": 42.0},
                  {"order_id": 1002, "region": "gb", "total": 17.5}):
        producer.produce(TOPIC, json.dumps(order, sort_keys=True).encode(), partition=PARTITION,
                         on_delivery=on_delivery)
    if producer.flush(10) != 0:
        raise RuntimeError("messages still in flight; the window is not closed")

    delivered.sort()
    start, end = delivered[0][0], delivered[-1][0] + 1  # [start, end)
    window = f"{TOPIC}-p{PARTITION}-{start}-{end}"
    run = root / window
    artifact = run / f"{window}.ndjson"
    run.mkdir(parents=True, exist_ok=True)
    artifact.write_bytes(window_bytes([value for _, value in delivered]))

    event = lambda name: f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{window}/{name}')}"
    head = mb.sign_origin(
        artifact, run / "attestations" / "01-window.dsse.json",
        source_kind="urn:example:source:kafka-offset-window",
        source_uri=f"kafka://{BROKER}/{TOPIC}",
        source_metadata={"version": f"partition={PARTITION};offsets={start}-{end}"},
        event_id=event("window"),
    )
    anchors = mb.create_handoff(
        head, artifact, run / "attestations", run / "bundle",
        recipient=RECIPIENT, bundle_id=event("bundle"), media_type="application/x-ndjson",
    )
    notice = {"topic": TOPIC, "partition": PARTITION, "start": start, "end": end,
              "bundle": str(run / "bundle"), "anchors": anchors}
    (run / "notice.json").write_text(json.dumps(notice))
    print(json.dumps({"window": window, "notice": str(run / "notice.json")}))


if __name__ == "__main__":
    produce(Path(sys.argv[1]).resolve())

3. The consumer

Save this as window_consumer.py.

"""Read a signed window from the topic and commit offsets only if Makoto allows it.

    python window_consumer.py /srv/makoto/windows/orders-p0-0-2/notice.json
"""

import copy
import hashlib
import json
import sys
from pathlib import Path

from confluent_kafka import Consumer, TopicPartition

import makoto_boundary as mb
from window_producer import BROKER, RECIPIENT, window_bytes

GROUP = "finance-orders"


def consume(notice_path: Path) -> int:
    notice = json.loads(notice_path.read_text())
    consumer = Consumer({
        "bootstrap.servers": BROKER,
        "group.id": GROUP,
        "enable.auto.commit": False,  # offsets move only after a verdict
        "auto.offset.reset": "earliest",
    })
    partition = TopicPartition(notice["topic"], notice["partition"], notice["start"])
    consumer.assign([partition])

    values: list[bytes] = []
    while len(values) < notice["end"] - notice["start"]:
        msg = consumer.poll(10)
        if msg is None:
            raise RuntimeError("window is incomplete on the broker")
        if msg.error():
            raise RuntimeError(msg.error())
        values.append(msg.value())

    # The expected artifact is what THIS consumer read, not what the bundle claims.
    anchors = copy.deepcopy(notice["anchors"])
    anchors["artifact"]["digest"]["sha256"] = hashlib.sha256(window_bytes(values)).hexdigest()
    try:
        mb.verify(notice["bundle"], anchors, recipient=RECIPIENT,
                  report_path=notice_path.parent / "verification.json")
    except mb.MakotoDenied as denied:
        consumer.close()  # nothing committed: the window will be read again, or quarantined
        print(f"REFUSED window {notice['start']}-{notice['end']}: {denied}", file=sys.stderr)
        return 1

    # Makoto proved these bytes are the signed bytes. It did not prove they came from
    # these offsets: identical records elsewhere in the log hash the same. Check the
    # signed window against the window this consumer actually read.
    head = notice["anchors"]["head"].removeprefix("sha256:")
    statement = json.loads(mb.payload(Path(notice["bundle"]) / "attestations" / f"{head}.dsse.json"))
    signed_window = statement["predicate"]["source"]["version"]
    read_window = f"partition={notice['partition']};offsets={notice['start']}-{notice['end']}"
    if signed_window != read_window:
        consumer.close()
        print(f"REFUSED window {notice['start']}-{notice['end']}: evidence signs {signed_window}", file=sys.stderr)
        return 1

    # Hand the verified records to the application here, then commit past the window.
    consumer.commit(offsets=[TopicPartition(notice["topic"], notice["partition"], notice["end"])],
                    asynchronous=False)
    consumer.close()
    print(f"COMMITTED window {notice['start']}-{notice['end']} ({len(values)} records)")
    return 0


if __name__ == "__main__":
    sys.exit(consume(Path(sys.argv[1]).resolve()))

4. Run it

python window_producer.py windows
python window_producer.py windows   # a second window with identical records
python window_consumer.py windows/orders-p0-0-2/notice.json

{"window": "orders-p0-0-2", "notice": "/private/tmp/claude-501/-Users-daaronch--treehouse-usemakoto-dev-e5b6d4-3-usemakoto-dev/7a8e8b40-d349-450c-be40-c39e2d154d0f/scratchpad/integ/kafka/windows/orders-p0-0-2/notice.json"}
COMMITTED window 0-2 (2 records)

5. What a gate failure looks like

Point the consumer at offsets the evidence does not cover. Shifted by one, the records come back in a different order, the bytes differ, and the verifier refuses:

jq '.start += 1 | .end += 1' notice.json > shift1.json
python window_consumer.py shift1.json
REFUSED window 1-3: Makoto deny (E_EXPECTED_ARTIFACT): step 11 E_EXPECTED_ARTIFACT: expected artifact set does not match

Shifted by two, the consumer reads the second window. Its records are byte-for-byte identical to the first, so the verifier allows the bytes. The window check refuses it:

jq '.start += 2 | .end += 2' notice.json > shift2.json
python window_consumer.py shift2.json
REFUSED window 2-4: evidence signs partition=0;offsets=0-2

In both cases no offset is committed, so the consumer group has not advanced past data it could not verify.

Bytes are not positions

The second refusal is the lesson of this page. Makoto authenticates exact bytes. Identical records at different offsets hash identically, so an artifact digest cannot tell a replay of the same values from the window that was signed. Whenever position is part of meaning, as it is for offsets, time windows, or partitions, put the position in signed metadata and compare it yourself after the verifier allows the bytes.

Other attachment points on this platform

Source window

An authorized finalizer closes declared partition and offset ranges, computes or resolves the exact window artifact, and creates an origin statement.

Stream transform

After an output window commits, append a transformation statement binding exact input window and predecessor digests.

Schema Registry

Carry schema IDs and serialization details as context or profile references. A schema ID does not become a Makoto predicate automatically.

Cross-team topic

Sign a handoff that commits to exact windows, heads, profiles, recipient, and final artifact representation.

What the evidence can say

It can bind one closed stream window to authorized claims and predecessors. It cannot prove device firmware, infer trust from a producer ID, or turn ongoing CDC into a complete history without an explicit finalization boundary.

Troubleshooting

Connect to ipv6#[::1]:9092 failed: Connection refused

librdkafka tried IPv6 first because the broker advertises localhost and the container published only an IPv4 port. It fell back to IPv4 and every run completed; the message is noise, not a failure.

RuntimeError: window is incomplete on the broker

Retention, compaction, or a truncated partition removed part of the window. The window can no longer be verified; quarantine it rather than committing past it.

Offsets were committed for a refused window

Auto-commit must be off. Commit explicitly, and only after both the verifier and the window check pass.