Snowflake: sign a stable export, not a mutable query result.

Where Makoto sits

Signing needs a key and a filesystem, so it happens in the process that drives Snowflake, not inside a stored procedure. Time Travel pins the snapshot so the export is reproducible.

1. Pin the snapshotRead CURRENT_TIMESTAMP() once and query the table AT that instant.
2. UnloadCOPY INTO @stage with SINGLE = TRUE and no compression writes one exact file.
3. Download and signGET the staged file and sign its bytes as an origin. The unload query ID is recorded as the source version.
4. Hand offThe bundle and anchors go to the receiving account.
5. Receiver gateVerify first. Only an allow reaches PUT and COPY INTO.

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 snowflake-connector-python

Both connections are read from ~/.snowflake/connections.toml by name, so no credential appears in the script.

uv run makoto_bootstrap.py /srv/makoto \
  --source-kind urn:example:source:snowflake-unload \
  --source-uri snowflake://acme/ANALYTICS.SALES.ORDERS \
  --operation-type urn:example:operation:none
export MAKOTO_KEYS=/srv/makoto/keys MAKOTO_POLICY=/srv/makoto/policy.json

2. The producer and receiver

Save this as snowflake_handoff.py.

"""Unload a Snowflake snapshot, sign the exact file, and gate the load on the receiver.

    python snowflake_handoff.py produce /srv/makoto/handoffs
    python snowflake_handoff.py receive /srv/makoto/handoffs/<run>
"""

import json
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import snowflake.connector

import makoto_boundary as mb

TABLE = "ANALYTICS.SALES.ORDERS"
RECIPIENT = "example:finance-account"


def produce(root: Path) -> Path:
    run = uuid.uuid4().hex
    local = (root / run).resolve()
    local.mkdir(parents=True)
    event = lambda name: f"urn:uuid:{uuid.uuid5(uuid.NAMESPACE_URL, f'{run}/{name}')}"

    with snowflake.connector.connect(connection_name="producer") as conn, conn.cursor() as cur:
        # Pin one snapshot: every statement below reads the table as of this instant.
        cur.execute("SELECT CURRENT_TIMESTAMP()::STRING")
        (as_of,) = cur.fetchone()
        cur.execute(
            f"""
            COPY INTO @ANALYTICS.SALES.HANDOFF_STAGE/{run}/orders.csv
            FROM (
              SELECT order_id, LOWER(TRIM(region)) AS region, total
              FROM {TABLE} AT(TIMESTAMP => '{as_of}'::TIMESTAMP_LTZ)
              ORDER BY order_id
            )
            FILE_FORMAT = (TYPE = CSV COMPRESSION = NONE FIELD_OPTIONALLY_ENCLOSED_BY = '"')
            HEADER = TRUE
            SINGLE = TRUE
            OVERWRITE = TRUE
            """
        )
        unload_query = cur.sfqid
        # GET downloads the exact staged bytes; hash those, never a re-query.
        cur.execute(f"GET @ANALYTICS.SALES.HANDOFF_STAGE/{run}/orders.csv 'file://{local}/'")

    export = local / "orders.csv"
    head = mb.sign_origin(
        export, local / "attestations" / "01-unload.dsse.json",
        source_kind="urn:example:source:snowflake-unload",
        source_uri=f"snowflake://acme/{TABLE}",
        source_metadata={
            "version": unload_query,
            "retrievedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "mediaType": "text/csv",
        },
        event_id=event("unload"),
    )
    anchors = mb.create_handoff(
        head, export, local / "attestations", local / "bundle",
        recipient=RECIPIENT, bundle_id=event("bundle"), media_type="text/csv",
    )
    # Publish anchors through a channel the receiver trusts independently of the bundle.
    (local / "anchors.json").write_text(json.dumps(anchors))
    return local


def receive(delivery: Path) -> None:
    anchors = json.loads((delivery / "anchors.json").read_text())
    try:
        mb.verify(delivery / "bundle", anchors, recipient=RECIPIENT,
                  report_path=delivery / "verification.json")
    except mb.MakotoDenied as denied:
        # Nothing has touched the warehouse yet. Refuse before PUT, not after COPY.
        raise SystemExit(f"refusing to load {delivery.name}: {denied}")

    final = next((delivery / "bundle" / "artifacts" / "final").glob("*.bin"))
    with snowflake.connector.connect(connection_name="receiver") as conn, conn.cursor() as cur:
        cur.execute(f"PUT 'file://{final}' @FINANCE.RAW.INBOUND/{delivery.name}/ AUTO_COMPRESS = FALSE")
        cur.execute(
            f"""
            COPY INTO FINANCE.RAW.ORDERS
            FROM @FINANCE.RAW.INBOUND/{delivery.name}/
            FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1 FIELD_OPTIONALLY_ENCLOSED_BY = '"')
            """
        )


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

3. Run it

python snowflake_handoff.py produce /srv/makoto/handoffs
# prints /srv/makoto/handoffs/<run>
python snowflake_handoff.py receive /srv/makoto/handoffs/<run>

4. What a gate failure looks like

If the delivered file was changed after signing, receive exits before connecting to the receiving account. From the stubbed run:

refusing to load 29d1d7ee882044eaa97a513ef65e7cfc: Makoto deny (E_ARTIFACT_DIGEST): step 12 E_ARTIFACT_DIGEST: final artifact bytes differ from the signed digest
warehouse calls after deny: []

Design notes

Why not sign inside Snowflake

A Snowpark procedure could hash staged files, but the signing key would then live in the warehouse’s trust boundary. Keep keys where the producer controls them.

Secure Data Sharing

A share exposes live objects, not bytes, so it cannot be verified as a handoff. Share the unloaded file set, or unload on the receiving side from the share and treat that as a new origin.

Several files

Without SINGLE = TRUE, an unload writes many files. Sign a dataset manifest over them instead of one file.

Other attachment points on this platform

Unload or export

After COPY INTO finishes, enumerate the exact stage objects, hash them outside the warehouse or through a reviewed signing service, and append a transformation statement.

Dynamic table or task

Capture query, task, warehouse, role, and snapshot context as operation metadata. Bind the exported bytes, not just a query ID.

Secure share

Create a signed handoff that names the exported artifact set, recipient, required profiles, and final heads.

Receiver

Download or read the handed-off bytes, hash them independently, and apply its own key policy and schema catalog.

What the evidence can say

It can bind a stable export and relevant Snowflake context to its predecessors. It cannot prove a query was semantically correct, that a share remained unchanged after the attested export, or that the data meets a compliance regime.

Troubleshooting

The same query produces different bytes on a rerun

The table changed. Query AT a pinned timestamp or statement, and add ORDER BY; unloaded row order is otherwise not guaranteed.

GET writes a compressed file

Set COMPRESSION = NONE in the unload file format, or sign the compressed bytes consistently on both sides.