Sulkta-Coop/cardano-checkout-py
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

v0.2: wire mint + txbuild end-to-end against local Ogmios

Browse source 
txbuild.py gets real: make_ogmios_context builds an
OgmiosChainContext pointed at 127.0.0.1:1337 by default (matches the
Rackham stack), get_protocol_parameters peeks at live params,
get_address_utxos powers the eventual refund path, submit_signed_tx
ships a cold-signed blob to the chain.

mint.py's mint_nft_cert now constructs the real tx body:
  - mints exactly 1 of {policy_id}.{asset_name}
  - sends it to recipient_address in its own UTxO padded with min-ADA
  - attaches CIP-25 v2 metadata + the native script as aux data
  - clamps TTL to policy.locked_after_slot when the policy is time-locked

Does NOT sign — matches the ADAMaps cold-signing pattern. Returns an
UnsignedMint bundle carrying body CBOR, aux CBOR, native-script CBOR,
required-signer hashes, and a human-readable summary. Operator moves
the bundle to the cold host (Lucy in Sulkta's pattern — 2-of-2
native-script under Cobb + Kayos skeys), signs offline, ships the
signed CBOR back. submit_signed_tx finishes the round-trip.
This commit is contained in:
Kayos 2026-04-23 19:55:45 -07:00
parent a6d4ac8521
commit 27b119bbe1
2 changed files with 523 additions and 68 deletions
Download patch file Download diff file Expand all files Collapse all files

376
cardano_checkout/mint.py
View file

@ -22,15 +22,39 @@ Design decisions:
CIP-25 metadata keeps the tx cheap (~0.18 ADA fee + min-utxo for the
NFT output).
v0.1.0 ships the signature + stub. Full tx construction against a local
Ogmios endpoint lands in v0.2 once the store + monitor integration
tests pass.
Cold-signing workflow
---------------------
The mint function does *not* sign. It builds the transaction body + the
auxiliary data, computes the tx id, and returns an :class:`UnsignedMint`
carrying the CBOR-encoded body plus a human-readable summary so the
operator can sanity-check before signing. The operator then:
1. Transfers the unsigned CBOR to the cold host (Lucy, via `scp`, USB,
QR code, whatever the threat model tolerates).
2. Signs offline with the policy-required skey(s) — for Sulkta's
chromatic policy that's ``Cobb.skey`` + ``Kayos.skey``.
3. Transfers the signed CBOR back to the hot host.
4. Calls :func:`submit_signed_tx` to hand it to Ogmios.
See ``docs/minting-workflow.md`` for the full operator runbook.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from typing import Optional
from typing import TYPE_CHECKING, Optional
logger = logging.getLogger(__name__)
if TYPE_CHECKING: # pragma: no cover — hints only
from pycardano import ChainContext
# ---------------------------------------------------------------------------
# Policy model
# ---------------------------------------------------------------------------
@dataclass
@ -43,10 +67,9 @@ class MintPolicy:
under it. Becomes the Cardano ``policy_id`` of the NFT asset.
script_cbor_hex: Hex-encoded CBOR of the native script itself.
Submitted alongside the mint tx witness.
signing_keys: Paths to skey files needed to sign the tx (all of
them, for an all-of multi-sig). The SDK does not read these —
callers pass them to :mod:`cardano_checkout.txbuild` which
coordinates with an external signer (Lucy cold-store pattern).
required_signer_hashes: Payment-key hashes (hex) of every skey
that must sign the mint tx. For Sulkta's chromatic policy
this is 2 entries: Cobb + Kayos.
locked_after_slot: Optional slot beyond which the policy rejects
further mints. None = no time lock (not recommended for
certificates — a lock makes the "no more editions" claim
@ -55,48 +78,43 @@ class MintPolicy:
policy_id: str
script_cbor_hex: str
signing_keys: list[str] = field(default_factory=list)
required_signer_hashes: list[str] = field(default_factory=list)
locked_after_slot: Optional[int] = None
async def mint_nft_cert(
policy: MintPolicy,
asset_name: str,
metadata: dict,
recipient_address: str,
ogmios_url: str = "http://127.0.0.1:1337",
network: str = "mainnet",
) -> str:
"""Mint a CIP-25 v2 NFT cert and send it to the recipient.
@dataclass
class UnsignedMint:
"""An unsigned mint transaction, ready to be handed to a cold signer.
Constructs a transaction that:
1. Mints exactly 1 of ``{policy.policy_id}.{asset_name}``.
2. Sends that single token to ``recipient_address`` in its own UTxO.
3. Attaches CIP-25 v2 metadata under metadatum label 721.
4. Witnesses with all signing keys required by the policy.
Args:
policy: Merchant's minting policy.
asset_name: UTF-8 asset name (will be hex-encoded per CIP-25). Max 32 bytes.
metadata: CIP-25 metadata dict. At minimum should include ``name``,
``image`` (ipfs://CID), ``mediaType``, and any studio-specific
properties. The SDK wraps this into the proper ``{721: {policy_id: {asset_name: ...}}}``
envelope automatically.
recipient_address: Bech32 address of the wallet that receives the NFT.
ogmios_url: Endpoint for chain queries + tx submission.
network: "mainnet" or "testnet".
Returns:
Transaction hash (hex) once successfully submitted.
Raises:
NotImplementedError: v0.1.0 stub. Full implementation lands in v0.2.
Attributes:
tx_id: Transaction hash computed from the body alone (stable across
signing — the same id the explorer will show once submitted).
tx_body_cbor_hex: Hex-encoded CBOR of the transaction *body*.
This is what gets moved to the cold host.
auxiliary_data_cbor_hex: Hex-encoded CBOR of the auxiliary data
(metadata + native script). Required to reconstruct the full
transaction before submission.
native_script_cbor_hex: Hex-encoded CBOR of the minting policy's
native script. Needed by the cold signer to construct the
correct witness set.
required_signer_hashes: List of payment-key hashes (hex) the cold
signer must provide. Mirrors ``MintPolicy.required_signer_hashes``.
summary: Human-readable description of the tx — operator should
eyeball this before signing to confirm they're signing what
they think they're signing.
"""
# v0.1.0 surface lock — implementation lands in v0.2 alongside txbuild.py
raise NotImplementedError(
"mint_nft_cert is stubbed in v0.1.0. Use cardano-cli or PyCardano "
"directly until v0.2 ships the full Ogmios-backed mint path."
)
tx_id: str
tx_body_cbor_hex: str
auxiliary_data_cbor_hex: str
native_script_cbor_hex: str
required_signer_hashes: list[str]
summary: str
# ---------------------------------------------------------------------------
# Metadata builder (pure, no pycardano dep)
# ---------------------------------------------------------------------------
def build_cip25_metadata(
@ -127,6 +145,7 @@ def build_cip25_metadata(
Returns:
Dict ready to submit as tx metadatum label 721.
"""
def chunk64(s: str) -> list[str]:
if len(s) <= 64:
return [s]
@ -154,3 +173,270 @@ def build_cip25_metadata(
"version": "2.0",
}
}
# ---------------------------------------------------------------------------
# Mint transaction builder (cold-signer flow)
# ---------------------------------------------------------------------------
def _require_pycardano():
try:
import pycardano # noqa: F401
except ImportError as exc: # pragma: no cover — env sanity
raise RuntimeError(
"pycardano is required for mint transaction construction. "
"Add pycardano>=0.11.0 to requirements.txt and reinstall."
) from exc
def _metadata_dict_with_int_keys(metadata: dict) -> dict:
"""Convert string top-level metadata labels to ints for pycardano Metadata.
CIP-25 v2 nests everything under label ``721``. We accept both ``{"721": ...}``
(builder output) and ``{721: ...}`` (raw) for ergonomics.
"""
converted: dict = {}
for key, val in metadata.items():
try:
converted[int(key)] = val
except (TypeError, ValueError):
converted[key] = val
return converted
async def mint_nft_cert(
policy: MintPolicy,
asset_name: str,
metadata: dict,
recipient_address: str,
funding_address: str,
context: Optional["ChainContext"] = None,
ogmios_host: str = "127.0.0.1",
ogmios_port: int = 1337,
network: str = "mainnet",
min_lovelace_for_nft_utxo: int = 1_500_000,
) -> UnsignedMint:
"""Build an unsigned mint+send transaction for a CIP-25 v2 NFT cert.
Constructs a transaction that:
1. Mints exactly 1 of ``{policy.policy_id}.{asset_name}``.
2. Sends that single token to ``recipient_address`` in its own UTxO
with the minimum-ADA padding (default 1.5 ADA).
3. Attaches the CIP-25 v2 metadata (label 721) + the policy's
native script as tx auxiliary data.
4. Returns the unsigned body for the cold signer to sign — does NOT
sign, does NOT submit.
UTxOs for fees + min-ADA are sourced from ``funding_address`` (the
merchant's hot wallet on Rackham, which does not hold any policy keys).
Args:
policy: Merchant's minting policy.
asset_name: UTF-8 asset name (will be hex-encoded per CIP-25). Max 32 bytes.
metadata: CIP-25 metadata dict — typically the output of
:func:`build_cip25_metadata`. Accepts ``{"721": ...}`` or ``{721: ...}``.
recipient_address: Bech32 address of the wallet that receives the NFT.
funding_address: Bech32 address that pays the tx fee + NFT min-ADA.
context: Optional chain context. If omitted a fresh
:class:`pycardano.OgmiosChainContext` is built from
``ogmios_host``/``ogmios_port``.
ogmios_host: Host of the local Ogmios HTTP+WS endpoint.
ogmios_port: Port of the local Ogmios endpoint.
network: ``"mainnet"`` or ``"testnet"`` (preprod / preview).
min_lovelace_for_nft_utxo: ADA (in lovelace) to attach to the NFT
output so it satisfies the ledger's min-UTxO floor. Default 1.5 ADA.
Returns:
:class:`UnsignedMint` bundle ready for the cold-signer hand-off.
Raises:
RuntimeError: If pycardano is unavailable, or tx construction fails.
ValueError: If ``asset_name`` is empty or > 32 bytes.
"""
_require_pycardano()
if not asset_name or len(asset_name.encode("utf-8")) > 32:
raise ValueError(
"asset_name must be a non-empty UTF-8 string <= 32 bytes "
f"(got {len(asset_name.encode('utf-8'))} bytes)"
)
from pycardano import (
Address,
Asset,
AssetName,
AuxiliaryData,
Metadata,
MultiAsset,
NativeScript,
Network,
ScriptHash,
TransactionBuilder,
TransactionOutput,
Value,
)
if context is None:
from cardano_checkout.txbuild import make_ogmios_context
context = make_ogmios_context(
host=ogmios_host, port=ogmios_port, network=network
)
net = Network.MAINNET if network == "mainnet" else Network.TESTNET
# ------------------------------------------------------------------
# Assemble the mint MultiAsset
# ------------------------------------------------------------------
policy_hash = ScriptHash.from_primitive(bytes.fromhex(policy.policy_id))
asset_name_obj = AssetName(asset_name.encode("utf-8"))
asset = Asset()
asset[asset_name_obj] = 1
mint_bundle = MultiAsset()
mint_bundle[policy_hash] = asset
# ------------------------------------------------------------------
# Native script + auxiliary data (metadata + script witness)
# ------------------------------------------------------------------
native_script = NativeScript.from_cbor(bytes.fromhex(policy.script_cbor_hex))
metadata_obj = Metadata(_metadata_dict_with_int_keys(metadata))
aux = AuxiliaryData(metadata_obj)
# AuxiliaryData in pycardano also carries native_scripts attached to the tx body;
# the builder below handles native scripts separately via add_minting_script.
# ------------------------------------------------------------------
# Addresses
# ------------------------------------------------------------------
sender = Address.from_primitive(funding_address)
recipient = Address.from_primitive(recipient_address)
if sender.network != net or recipient.network != net:
raise ValueError(
f"Address network mismatch: requested {network}, "
f"sender={sender.network.name}, recipient={recipient.network.name}"
)
# ------------------------------------------------------------------
# Build the transaction
# ------------------------------------------------------------------
builder = TransactionBuilder(context)
builder.add_input_address(sender)
# Attach mint bundle + policy as a minting script.
builder.mint = mint_bundle
builder.native_scripts = [native_script]
builder.auxiliary_data = aux
# Output: the newly minted NFT in its own UTxO at the recipient, padded
# with min-ADA so the ledger accepts it.
nft_value = Value(min_lovelace_for_nft_utxo, mint_bundle)
builder.add_output(TransactionOutput(recipient, nft_value))
# If the policy has a time lock, the mint tx MUST set ttl <= locked_after_slot
# or the node will reject the witness. Let pycardano pick validity normally,
# but clamp ttl when a lock slot is set.
ttl_offset = None
if policy.locked_after_slot is not None:
try:
chain_tip = context.last_block_slot # type: ignore[attr-defined]
# Cap at 2 hours or (locked_after_slot - chain_tip), whichever is smaller.
two_hours_in_slots = 2 * 60 * 60 # ~1 slot/s on mainnet
ttl_offset = max(
60, min(two_hours_in_slots, policy.locked_after_slot - chain_tip)
)
except Exception: # pragma: no cover — context without chain tip
ttl_offset = None
try:
tx_body = builder.build(
change_address=sender,
auto_ttl_offset=ttl_offset,
auto_validity_start_offset=-30,
)
except Exception as exc:
raise RuntimeError(f"Failed to build mint tx body: {exc}") from exc
tx_id = str(tx_body.id)
summary_lines = [
f"Mint 1 x {policy.policy_id}.{asset_name}",
f" -> recipient: {recipient_address}",
f" fees paid by: {funding_address}",
f" tx_id (pre-sign): {tx_id}",
f" network: {network}",
f" required signers: {len(policy.required_signer_hashes)} "
f"({', '.join(h[:16] + '...' for h in policy.required_signer_hashes) or 'NONE — check policy'})",
]
if policy.locked_after_slot is not None:
summary_lines.append(
f" policy time-lock: slot <= {policy.locked_after_slot}"
)
return UnsignedMint(
tx_id=tx_id,
tx_body_cbor_hex=tx_body.to_cbor_hex(),
auxiliary_data_cbor_hex=aux.to_cbor_hex(),
native_script_cbor_hex=policy.script_cbor_hex,
required_signer_hashes=list(policy.required_signer_hashes),
summary="\n".join(summary_lines),
)
# ---------------------------------------------------------------------------
# Signed-tx submission
# ---------------------------------------------------------------------------
def submit_signed_tx(
signed_tx_cbor_hex: str,
context: Optional["ChainContext"] = None,
ogmios_host: str = "127.0.0.1",
ogmios_port: int = 1337,
network: str = "mainnet",
) -> str:
"""Submit a cold-signed transaction to the network via Ogmios.
The cold signer produces a fully-assembled :class:`pycardano.Transaction`
— body + witness set + auxiliary data — serialised as CBOR. This
function deserialises that blob, hands it to Ogmios, and returns the
tx hash.
Args:
signed_tx_cbor_hex: Hex-encoded CBOR of the signed transaction.
context: Optional chain context; built from ``ogmios_host/port`` if omitted.
ogmios_host: Host of the Ogmios endpoint.
ogmios_port: Port of the Ogmios endpoint.
network: ``"mainnet"`` or ``"testnet"``.
Returns:
Transaction hash (hex) — stable identifier for the submitted tx.
Raises:
RuntimeError: If pycardano is unavailable, or submission fails.
"""
_require_pycardano()
from pycardano import Transaction
if context is None:
from cardano_checkout.txbuild import make_ogmios_context
context = make_ogmios_context(
host=ogmios_host, port=ogmios_port, network=network
)
try:
tx = Transaction.from_cbor(bytes.fromhex(signed_tx_cbor_hex))
except Exception as exc:
raise RuntimeError(f"signed_tx_cbor_hex is not valid transaction CBOR: {exc}") from exc
try:
context.submit_tx(tx) # type: ignore[attr-defined]
except Exception as exc:
raise RuntimeError(f"Ogmios rejected the signed tx: {exc}") from exc
tx_hash = str(tx.id)
logger.info("[mint] submitted signed tx %s", tx_hash)
return tx_hash

215
cardano_checkout/txbuild.py
View file

@ -1,38 +1,207 @@
"""Transaction construction helpers wrapping PyCardano.
v0.1.0 surface stub — the chromaticcraft Phase-2 sprint will fill these in
with:
- Ogmios-backed `ChainContext` (via PyCardano's OgmiosChainContext)
- Build-transaction helpers for (a) plain ADA payment refunds, (b)
native-token mint+send, (c) reference-asset clones
- Cold-signer hand-off shape matching the ADAMaps payout pattern:
build_body_on_hot → transfer via temp dir → sign_offline_on_cold →
return signed_witness → submit_from_hot.
This module is the SDK's single point of contact with PyCardano's
:class:`pycardano.backend.base.ChainContext` API. Everything higher up
(``mint`` and eventual refund-path code) goes through the helpers
here so we can swap Ogmios for Blockfrost / Cardano-CLI without
touching callers.
Exists as a named module in v0.1 so consumers can import the stable path
without having to update imports later.
The default context targets the local Ogmios instance on Rackham
(``127.0.0.1:1337``). That lines up with the mainnet deployment of the
``cardano-node`` container (v10.6.2 on port 6000 via N2N) fronted by
Ogmios as the HTTP+WS bridge. Preprod / testnet callers pass
``network="testnet"`` and typically point at a different host.
Cold-signer shape
-----------------
``txbuild`` only knows the hot-side half of the dance:
- :func:`make_ogmios_context` — build a context from the live node.
- :func:`get_protocol_parameters` — peek at the current protocol params
(useful for pricing, ttl calculations, etc.).
- :func:`get_address_utxos` — list UTxOs at an address (refund path).
- :func:`submit_signed_tx` — ship a tx that was signed offline.
Body construction lives in :mod:`cardano_checkout.mint` today. As
additional tx shapes (refunds, batched mints) arrive they'll land here
alongside ``build_*_tx`` helpers that return :class:`UnsignedMint`-style
cold-signer bundles.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Optional
def _v0_2_sentinel(_name: str) -> None:
raise NotImplementedError(
f"txbuild.{_name} ships in cardano-checkout v0.2 alongside the "
"Ogmios chain-context wiring and cold-signer hand-off."
logger = logging.getLogger(__name__)
if TYPE_CHECKING: # pragma: no cover — hints only
from pycardano import ChainContext, UTxO
# ---------------------------------------------------------------------------
# Chain context
# ---------------------------------------------------------------------------
def _require_pycardano() -> None:
try:
import pycardano # noqa: F401
except ImportError as exc: # pragma: no cover — env sanity
raise RuntimeError(
"pycardano is required for transaction construction. "
"Add pycardano>=0.11.0 to requirements.txt and reinstall."
) from exc
def make_ogmios_context(
host: str = "127.0.0.1",
port: int = 1337,
network: str = "mainnet",
secure: bool = False,
**kwargs: Any,
) -> "ChainContext":
"""Construct an :class:`pycardano.OgmiosChainContext` for the live node.
Args:
host: Ogmios HTTP+WS host. Default ``127.0.0.1`` (local).
port: Ogmios port. Default ``1337`` (matches Rackham's stack).
network: ``"mainnet"`` or ``"testnet"``. Controls the
:class:`pycardano.Network` passed to the context.
secure: Whether to use wss:// instead of ws://. Default False —
the stack assumes a loopback connection.
**kwargs: Forwarded to ``OgmiosChainContext`` verbatim (e.g.
``refetch_chain_tip_interval``, ``utxo_cache_size``).
Returns:
A live :class:`ChainContext`. If the backing node is down the
object is still constructed — failures surface on the first
query / submit call.
"""
_require_pycardano()
from pycardano import Network, OgmiosChainContext
net = Network.MAINNET if network == "mainnet" else Network.TESTNET
logger.debug(
"[txbuild] OgmiosChainContext -> %s://%s:%d (network=%s)",
"wss" if secure else "ws",
host,
port,
network,
)
return OgmiosChainContext(
host=host, port=port, secure=secure, network=net, **kwargs
)
def build_mint_tx(*args, **kwargs): # noqa: D401, ANN002, ANN003
"""Build an unsigned mint transaction. v0.2."""
_v0_2_sentinel("build_mint_tx")
def get_protocol_parameters(context: "ChainContext") -> Any:
"""Return the live protocol parameters from the chain context.
Useful for fee estimation, min-utxo floor computation, and sanity
checks that the node is reachable before a mint attempt.
The return type is pycardano's :class:`ProtocolParameters` — a
dataclass with fields like ``min_fee_a``, ``min_fee_b``,
``coins_per_utxo_byte``, ``max_tx_size``, etc.
"""
try:
return context.protocol_param # type: ignore[attr-defined]
except Exception as exc:
raise RuntimeError(
f"Failed to fetch protocol parameters from chain context: {exc}"
) from exc
def build_payment_tx(*args, **kwargs): # noqa: D401, ANN002, ANN003
"""Build an unsigned payment transaction (e.g. refund path). v0.2."""
_v0_2_sentinel("build_payment_tx")
def get_address_utxos(context: "ChainContext", address: str) -> list["UTxO"]:
"""Fetch UTxOs at ``address`` via the chain context.
Intended for the refund path — when an invoice is cancelled or
overpaid the merchant needs to know which UTxOs landed in order to
build a return tx. For pure payment-detection, Koios is still the
cheaper source (see :mod:`cardano_checkout.monitor`).
Args:
context: Live chain context (from :func:`make_ogmios_context`).
address: Bech32 Cardano address.
Returns:
List of pycardano :class:`UTxO` objects at ``address``. Empty if
the address has no unspent outputs. Never ``None``.
Raises:
RuntimeError: If the underlying query fails (node down, invalid address).
"""
_require_pycardano()
from pycardano import Address
try:
addr_obj = Address.from_primitive(address)
except Exception as exc:
raise RuntimeError(f"Invalid Cardano address: {exc}") from exc
try:
utxos = context.utxos(str(addr_obj)) # type: ignore[attr-defined]
except Exception as exc:
raise RuntimeError(
f"Failed to fetch UTxOs for {address[:20]}...: {exc}"
) from exc
return list(utxos or [])
def submit_signed_tx(*args, **kwargs): # noqa: D401, ANN002, ANN003
"""Submit a signed tx to Ogmios. v0.2."""
_v0_2_sentinel("submit_signed_tx")
# ---------------------------------------------------------------------------
# Signed-tx submission (duplicated from mint.py as a stable txbuild entry
# point — the mint module's version delegates here)
# ---------------------------------------------------------------------------
def submit_signed_tx(
signed_tx_cbor_hex: str,
context: Optional["ChainContext"] = None,
ogmios_host: str = "127.0.0.1",
ogmios_port: int = 1337,
network: str = "mainnet",
) -> str:
"""Submit a cold-signed transaction blob to the chain.
See :func:`cardano_checkout.mint.submit_signed_tx` for the full docstring —
this is the same function under the ``txbuild`` import path so callers
that only need submission don't have to import ``mint``.
"""
_require_pycardano()
from pycardano import Transaction
if context is None:
context = make_ogmios_context(
host=ogmios_host, port=ogmios_port, network=network
)
try:
tx = Transaction.from_cbor(bytes.fromhex(signed_tx_cbor_hex))
except Exception as exc:
raise RuntimeError(
f"signed_tx_cbor_hex is not valid transaction CBOR: {exc}"
) from exc
try:
context.submit_tx(tx) # type: ignore[attr-defined]
except Exception as exc:
raise RuntimeError(f"Ogmios rejected the signed tx: {exc}") from exc
tx_hash = str(tx.id)
logger.info("[txbuild] submitted signed tx %s", tx_hash)
return tx_hash
# ---------------------------------------------------------------------------
# Placeholders for future tx shapes (kept so consumers can pin imports)
# ---------------------------------------------------------------------------
def build_payment_tx(*args, **kwargs): # pragma: no cover — future work
"""Build an unsigned plain-ADA payment tx (refund path). v0.3+."""
raise NotImplementedError(
"build_payment_tx lands in v0.3 alongside the refund workflow. "
"For v0.2 only mint txs are supported."
)