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

This commit is contained in:
Sulkta 2026-04-23 19:55:45 -07:00
parent fb1e1b73a6
commit 3225e830a3
2 changed files with 523 additions and 68 deletions

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 (the cold host, via `scp`, USB,
QR code, whatever the threat model tolerates).
2. Signs offline with the policy-required skey(s) for Sulkta's
example policy that's ``signer1.skey`` + ``signer2.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 (the cold host cold-store pattern).
required_signer_hashes: Payment-key hashes (hex) of every skey
that must sign the mint tx. For Sulkta's example policy
this is 2 entries: signer 1 + signer 2.
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 the hot host, 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