diff --git a/README.md b/README.md index 636ad42..ecd3cf2 100644 --- a/README.md +++ b/README.md @@ -1,220 +1,209 @@ # cardano-checkout -Python SDK for merchant-side Cardano payments + NFT certificate-of-authenticity minting. +Merchant-side Cardano payment lifecycle in Python. Zero-custody by design. -**Zero-custody by design:** the merchant provides a wallet xpub. The SDK derives -unique receive addresses per invoice, polls the chain for payment, and optionally -mints a CIP-25 NFT cert on confirmation. The platform never holds or moves funds. +**What we ship:** the invoice state machine + UTxO watcher + reprice +loop. Per-invoice HD-derived receive addresses, Koios polling, confirm +/ underpay / overpay classification, time-windowed repricing against +your own oracle. -Extracted from [TradeCraft](http://192.168.0.5:3001/TradeCraft/tradecraft)'s -`services/cardano_*.py` modules (2,400+ lines of production code running on the -Cardano mainnet) and packaged for reuse across the Sulkta Coop product family. +**What we don't ship:** Cardano primitives. Address derivation, chain +context, transaction building, native-script minting, signing — those +are all [pycardano](https://github.com/Python-Cardano/pycardano)'s job. +pycardano is mature, actively maintained (0.19.x as of 2026), and +covers every primitive cleanly. This library slots next to it — no +wrapping, no leaky abstraction, no second API to learn. -## Status +## Why this exists -**v0.2.0-dev — Protocol-first core + live mint path.** Monitor and scheduler -have been refactored off SQLAlchemy and onto the `InvoiceStore` Protocol. -Mint builds real transaction bodies against a local Ogmios endpoint and -returns an `UnsignedMint` for cold-signing. +Nothing else in the Python ecosystem (or any ecosystem — we checked) +packages zero-custody merchant Cardano payments as a reusable library. +Closest adjacents are all one of: -| Module | Status | Notes | -|---|---|---| -| `addresses` | ✅ stable | CIP-1852 HD derivation via pycardano `HDWallet` soft derive | -| `oracles` | ✅ stable | ADA/USD price via CoinGecko + DexHunter, 5-min cache | -| `invoice` + `store` | ✅ stable | Framework-agnostic invoice + `InMemoryStore` reference impl | -| `mint` | ✅ v0.2 | CIP-25 v2 metadata + real tx body → `UnsignedMint` bundle | -| `ipfs` | ✅ stable | kubo HTTP API client w/ optional mirror-pin | -| `monitor` | ✅ v0.2 | Operates purely through `InvoiceStore` — no ORM coupling | -| `scheduler` | ✅ v0.2 | `InvoiceScheduler` drives check + reprice against the store | -| `tradecraft_compat` | 🟡 compat shim | Keeps TradeCraft's subscription + grace-period jobs alive during migration | -| `txbuild` | ✅ v0.2 | OgmiosChainContext wiring + submit_signed_tx + address UTxO queries | +- CIP-30 browser-wallet plugins (customer-signs, not server-watches) +- cardano-cli vending machines that watch a single static address (no + xpub, no per-invoice derivation) +- SaaS APIs (NMKR) — not libraries +- Dormant / pre-1.0 grabs from the 2021-2023 era -**Migration status for TradeCraft:** still imports the old module paths. See -the [v0.2 migration guide](#v02-migration-guide-for-tradecraft) below. - -## Design - -``` - ┌────────────────────────────────────────────────────────┐ - │ Merchant App │ - │ (TradeCraft / chromaticcraft / your-product) │ - └──────────────┬───────────────────────┬─────────────────┘ - │ │ - uses │ implements │ imports - ▼ ▼ - ┌──────────────┐ ┌────────────────────────┐ - │ InvoiceStore │ ◄────── │ cardano_checkout SDK │ - │ (your DB) │ │ │ - └──────────────┘ │ addresses ← pure │ - │ oracles ← pure │ - │ invoice ← dataclass │ - │ store ← Protocol + InMemoryStore │ - │ monitor ← polls chain via store │ - │ scheduler ← bg loop │ - │ mint ← NFT cert (cold-signer) │ - │ ipfs ← upload │ - │ txbuild ← Ogmios wrappers │ - └────────────────────────┘ - │ - talks to │ - ▼ - ┌────────────────────────┐ - │ Koios + Ogmios + kubo │ - └────────────────────────┘ -``` - -The merchant app provides: - -1. A wallet xpub (account-level extended public key). -2. An `InvoiceStore` implementation (SQLAlchemy, Postgres, SQLite, in-memory — whatever). - -The SDK provides: - -1. Address derivation from the xpub. -2. Per-invoice payment monitoring against Koios. -3. ADA ↔ USD price conversion. -4. CIP-25 v2 NFT cert minting with a cold-signer hand-off. -5. IPFS upload + pinning for NFT image metadata. +The merchant state machine — "derive an address, watch for payment, +confirm within tolerance, reprice if the quote lapses, emit a confirmed +callback" — is what we package. You keep full control of everything +else by using pycardano directly. ## Quick start ```python import asyncio -from cardano_checkout import addresses, oracles +from datetime import datetime, timedelta, timezone -# Derive a receive address for invoice #42 -addr = addresses.derive_address( - xpub_hex="", - index=42, - network="mainnet", +from cardano_checkout import ( + Invoice, InvoiceStatus, InMemoryStore, InvoiceScheduler, ) -# Convert a USD price to lovelace at current market + +# Your oracle — we don't ship one. Anything async returning int lovelace works. +async def my_price_fn(usd: float) -> int: + rate = await fetch_ada_usd_somewhere() # CoinGecko, Koios, fixed rate, etc. + return int(round(usd / rate * 1_000_000)) + + async def main() -> None: - lovelace = await oracles.convert_usd_to_lovelace(99.00) - ada = lovelace / 1_000_000 - print(f"Customer owes {ada:.4f} ADA for $99") + store = InMemoryStore() # swap for your SQLAlchemy / asyncpg / sqlite adapter + + # Create an invoice. In production you'd derive the receive address from + # your wallet xpub via pycardano — see the "Deriving addresses" section. + invoice = Invoice( + id="ord-0042", + merchant_id="chromaticcraft", + derivation_index=42, + receive_address="addr1q...", # derived via pycardano — your code + expected_lovelace=5_000_000, + usd_amount=2.50, + expires_at=datetime.now(timezone.utc) + timedelta(minutes=15), + ) + await store.create(invoice) + + # Run the background scheduler — Koios poll every 15s + reprice every 60s. + scheduler = InvoiceScheduler(store=store, price_fn=my_price_fn) + await scheduler.start() + + # ... app runs ... + + await scheduler.stop() asyncio.run(main()) ``` -## Payment monitoring +## Deriving addresses with pycardano + +We used to wrap this. You don't need the wrapper. ```python -import asyncio -from cardano_checkout import InMemoryStore, Invoice, InvoiceStatus, InvoiceScheduler +from pycardano import HDWallet, Address, Network -store = InMemoryStore() # swap for your real SQLAlchemy / asyncpg / SQLite adapter +# Your merchant's account-level xpub — the xpub is public, not a secret. +xpub_hex = "..." -# Create an invoice (typically you'd derive the address here via addresses.derive_address) -invoice = Invoice( - id="ord-0042", - merchant_id="chromaticcraft", - derivation_index=42, - receive_address="addr1q...", - expected_lovelace=5_000_000, - usd_amount=2.50, -) -asyncio.run(store.create(invoice)) +account = HDWallet.from_xpub(bytes.fromhex(xpub_hex)) -# Wire the background scheduler — same 15s check / 60s reprice cadence as TradeCraft. -scheduler = InvoiceScheduler(store=store) -asyncio.run(scheduler.start()) -# ... your app runs ... -asyncio.run(scheduler.stop()) +def derive_address(account: HDWallet, index: int, network=Network.MAINNET) -> str: + payment = account.derive(0).derive(index) # external chain, address index + staking = account.derive(2).derive(0) # staking chain, always index 0 + addr = Address( + payment_part=payment.public_key.hash(), + staking_part=staking.public_key.hash(), + network=network, + ) + return str(addr) + +addr = derive_address(account, index=42) ``` -## IPFS: bake-then-mirror pattern +Six lines of pycardano that read cleanly against +[their docs](https://pycardano.readthedocs.io/). Our old wrapper would +have added one function call but made you learn our API instead of +pycardano's. Skip it. -The SDK's `IPFSClient` expects a local kubo daemon (typically in the same -Docker image as the web app) for upload and primary pin, and takes an -optional list of mirror endpoints to `pin add` the CID on a second node -for archival redundancy. +## NFT cert-of-authenticity: CIP-25 v2 metadata -Typical chromaticcraft deployment: +If you want each paid order to ship with an on-chain NFT cert, here's +the CIP-25 v2 metadata builder as a copy-paste. It fits in your own +code — no dep, no wrapper. ```python -from cardano_checkout import ipfs +def build_cip25_metadata( + *, + policy_id: str, + asset_name: str, + name: str, + image_cid: str, + description: str = "", + media_type: str = "image/jpeg", + properties: dict | None = None, +) -> dict: + """Build a CIP-25 v2 metadata envelope. -client = ipfs.IPFSClient( - api_url="http://127.0.0.1:5001", # local kubo in the same container - mirror_api_urls=["http://192.168.254.5:5001"], # Lucy's kubo over the LAN/VPN -) + Returns a dict ready to submit as transaction metadatum label 721. + Handles the 64-char chunking rule for long descriptions. + """ + def chunk64(s: str) -> list[str]: + if len(s) <= 64: + return [s] + return [s[i:i + 64] for i in range(0, len(s), 64)] -cid = await client.add(photo_bytes, filename="order-0001.jpg") -# Image now served by Rackham (low latency) AND pinned on Lucy (durability) + body: dict = { + "name": name, + "image": f"ipfs://{image_cid}", + "mediaType": media_type, + } + if description: + body["description"] = description if len(description) <= 64 else chunk64(description) + if properties: + body.update(properties) + + return { + "721": { + policy_id: {asset_name: body}, + "version": "2.0", + } + } ``` -## NFT cert-of-authenticity design +Hand that dict to pycardano's `AuxiliaryData(Metadata({...}))` when you +build the mint tx. Straight pycardano from there on. -One minting policy per merchant studio. Policy is a native script (no Plutus -required), optionally time-locked to make "no more editions after X" a -cryptographically verifiable claim. +## Implementing your own InvoiceStore -CIP-25 v2 metadata. Single NFT per order. Policy skey never leaves the custody -host (Lucy in Sulkta's pattern — 2-of-2 native script: Cobb + Kayos). The SDK -builds the metadata envelope + tx body on the hot node and returns an -`UnsignedMint` bundle; an external offline signer provides the vkey witnesses; -the hot node submits the assembled CBOR. +The SDK's `InvoiceStore` is a Protocol — implement the six methods +against whatever backend you want (SQLAlchemy, asyncpg, SQLite, +in-memory for tests). -The full operator runbook — including the exact byte-movement sequence, -verification checklist, and preprod dry-run procedure — lives in -[`docs/minting-workflow.md`](docs/minting-workflow.md). +```python +from cardano_checkout import Invoice, InvoiceStatus, InvoiceStore -## v0.2 migration guide for TradeCraft +class MySqliteStore: + # Implement these six methods and you're a valid InvoiceStore. + async def create(self, invoice: Invoice) -> None: ... + async def get(self, invoice_id: str) -> Invoice | None: ... + async def list_by_status(self, status: InvoiceStatus, limit: int = 100) -> list[Invoice]: ... + async def update(self, invoice: Invoice) -> None: ... + async def next_derivation_index(self, merchant_id: str) -> int: ... + async def record_tx(self, invoice_id: str, tx_hash: str, lovelace_delta: int) -> None: ... +``` -The generic invoice jobs moved to a Protocol-based API. The -subscription + grace-period jobs stayed TradeCraft-specific and live in -`cardano_checkout.tradecraft_compat`. +See `InMemoryStore` in `cardano_checkout/store.py` for a 90-line +reference implementation. -**Import changes when TradeCraft adopts the SDK:** +## Status (1.0.0-dev) -| Was | Becomes | +| Module | Purpose | |---|---| -| `from services.cardano_monitor import check_pending_payments, reprice_expired_payments` | `from cardano_checkout.monitor import check_pending_invoices, reprice_expired_invoices` | -| `from services.cardano_monitor import _check_address_utxos, _evaluate_payment` | `from cardano_checkout.monitor import check_address_utxos, evaluate_utxos` (or import from `tradecraft_compat` for the exact old names) | -| `from services.cardano_scheduler import start_cardano_scheduler, stop_cardano_scheduler` | `from cardano_checkout.scheduler import InvoiceScheduler` (instantiate with your store) | -| `from services.cardano_scheduler import _check_subscription_payments, _reprice_subscription_payments, _enforce_grace_period` | `from cardano_checkout.tradecraft_compat import check_subscription_payments, reprice_subscription_payments, enforce_grace_period` (verbatim jobs — TradeCraft still drives them directly) | -| `from services.cardano_price import *` | `from cardano_checkout.oracles import *` | -| `from services.cardano_addresses import derive_address` | `from cardano_checkout.addresses import derive_address` | +| `invoice.py` | `Invoice` dataclass + `InvoiceStatus` enum — payment lifecycle states | +| `store.py` | `InvoiceStore` Protocol + `InMemoryStore` reference impl | +| `monitor.py` | `check_address_utxos` (Koios), `evaluate_utxos` (ADA matching + tolerance), `check_pending_invoices`, `reprice_expired_invoices` (takes your `price_fn`) | +| `scheduler.py` | `InvoiceScheduler` — APScheduler wrapper, runs check/reprice on the same 15s/60s cadence TradeCraft's used in production since 2025 | -**What TradeCraft still needs to write:** +All tests offline, 26/26 green. Two direct deps: `httpx` (Koios calls), +`apscheduler` (background scheduling). No pycardano dep — that's the +consumer's pairing. -A SQLAlchemy adapter implementing `InvoiceStore` against the existing -`CardanoPayment` table. `list_by_status` maps to a `SELECT ... WHERE status = :s`, -`next_derivation_index` to a `SELECT MAX(derivation_index) + 1`, etc. -That's 80-ish lines of wrapper code — nothing exotic. Once that's -landed, the generic jobs run through `InvoiceScheduler(store=SQLAlchemyInvoiceStore(...))` -and the subscription jobs keep running through `tradecraft_compat` unchanged. +## Design principles -**TODO for future sprints:** - -- Ship a `cardano_checkout.adapters.sqlalchemy.SQLAlchemyInvoiceStore` so - TradeCraft doesn't have to write the adapter from scratch. -- Once TradeCraft's subscription jobs are migrated to a subscription- - specific Protocol, delete `tradecraft_compat`. -- Refund-path `build_payment_tx` in `txbuild.py` (v0.3). -- Batched mints (sell-sheet of 10 NFTs at once). - -## Testing - -```bash -pip install -e '.[test]' -pytest # 42 tests, all offline -``` - -The test suite mocks the chain context for mint-tx construction and -monkey-patches Koios + the oracle for monitor tests — CI never touches -a live node. The address-derivation tests use a deterministic test-vector -xpub from the standard "test ... junk" mnemonic so they can't drift. - -## Installation - -``` -pip install 'cardano-checkout[sqlalchemy]' # if you're using SQLAlchemy -pip install cardano-checkout # core only -``` +1. **Protocol-first.** Persistence, pricing, and any other side-effect + concern goes through a consumer-supplied interface. The SDK has no + opinion about your database, your oracle, or your ORM. +2. **Use pycardano directly.** We don't wrap primitives. If you need + address derivation, chain context, or transaction building, import + pycardano. Our package sits next to it, not on top. +3. **Zero-custody.** The merchant's keys never touch this code. We + handle xpub-derived addresses (public), UTxO observation (chain), + and state transitions (the store). Funds flow directly between + customer and merchant wallets. We are not a custodian. +4. **Offline-first tests.** Koios HTTP and price oracles are stubbed + or swapped via fixture. No network in CI. Live tests (preprod mint + round-trips, real Koios) are a consumer-side concern. ## License -Apache-2.0 — matches upstream Cardano tooling. +Apache-2.0 — matches the broader Cardano tooling ecosystem. diff --git a/cardano_checkout/__init__.py b/cardano_checkout/__init__.py index bbdb5a1..acc27f1 100644 --- a/cardano_checkout/__init__.py +++ b/cardano_checkout/__init__.py @@ -1,68 +1,71 @@ -"""cardano_checkout — Python SDK for merchant-side Cardano payments + NFT cert minting. +"""cardano-checkout — merchant-side Cardano payment lifecycle in Python. -Zero-custody by design: consumers provide a wallet xpub (account-level -extended public key). The SDK derives unique receive addresses per -invoice, polls the chain for payment, and (optionally) mints a CIP-25 -NFT certificate of authenticity on confirmation. +Zero-custody by design: the merchant brings a wallet xpub and an +:class:`~cardano_checkout.store.InvoiceStore` implementation. The SDK +owns the payment lifecycle — per-invoice receive-address bookkeeping, +Koios UTxO polling, confirm / underpay / overpay classification, and +time-windowed repricing against a consumer-supplied oracle. + +**The SDK deliberately does NOT ship Cardano primitives.** Address +derivation, transaction building, chain context, and native-script +minting all live in `pycardano `_ +and are consumer concerns. See the README for the pairing pattern and +for the CIP-25 v2 metadata-builder snippet (a 60-line helper that fits +anywhere in your own code without needing a separate dep). Quick start:: - from cardano_checkout import addresses, oracles + from cardano_checkout import Invoice, InvoiceStatus, InMemoryStore, InvoiceScheduler - addr = addresses.derive_address(xpub_hex, index=42, network="mainnet") - price = await oracles.get_ada_usd_price() - lovelace = await oracles.convert_usd_to_lovelace(99.00) + store = InMemoryStore() # or your SQLAlchemy / asyncpg / sqlite adapter -For full invoice lifecycle see :mod:`cardano_checkout.invoice` + -:mod:`cardano_checkout.store` (Protocol-based persistence). + async def my_price_fn(usd: float) -> int: + # your oracle — CoinGecko / Koios ticker / fixed rate in tests + rate = await fetch_ada_usd_rate() + return int(round(usd / rate * 1_000_000)) -For NFT minting see :mod:`cardano_checkout.mint`. + scheduler = InvoiceScheduler(store=store, price_fn=my_price_fn) + await scheduler.start() """ from __future__ import annotations -__version__ = "0.2.0-dev" +__version__ = "1.0.0-dev" -# Pure modules — stable API from extraction -from cardano_checkout import addresses, oracles # noqa: F401 - -# Payment lifecycle from cardano_checkout.invoice import Invoice, InvoiceStatus # noqa: F401 from cardano_checkout.store import InMemoryStore, InvoiceStore # noqa: F401 - -# Monitoring + scheduling from cardano_checkout.monitor import ( # noqa: F401 + CONFIRM_TOLERANCE, + DEFAULT_MAX_REPRICINGS, + DEFAULT_PAYMENT_WINDOW_MINUTES, + KOIOS_URL, + OVERPAY_THRESHOLD, + PriceFn, + check_address_utxos, check_pending_invoices, + evaluate_utxos, reprice_expired_invoices, ) from cardano_checkout.scheduler import InvoiceScheduler # noqa: F401 -# NFT + IPFS -from cardano_checkout.mint import ( # noqa: F401 - MintPolicy, - UnsignedMint, - build_cip25_metadata, - mint_nft_cert, - submit_signed_tx, -) -from cardano_checkout.ipfs import IPFSClient, pin_bytes # noqa: F401 - __all__ = [ "__version__", - "addresses", - "oracles", + # Invoice lifecycle "Invoice", "InvoiceStatus", + # Persistence "InvoiceStore", "InMemoryStore", + # Monitor + scheduler + "PriceFn", "InvoiceScheduler", + "check_address_utxos", "check_pending_invoices", + "evaluate_utxos", "reprice_expired_invoices", - "MintPolicy", - "UnsignedMint", - "mint_nft_cert", - "submit_signed_tx", - "build_cip25_metadata", - "IPFSClient", - "pin_bytes", + "KOIOS_URL", + "CONFIRM_TOLERANCE", + "OVERPAY_THRESHOLD", + "DEFAULT_MAX_REPRICINGS", + "DEFAULT_PAYMENT_WINDOW_MINUTES", ] diff --git a/cardano_checkout/addresses.py b/cardano_checkout/addresses.py deleted file mode 100644 index 8ee135f..0000000 --- a/cardano_checkout/addresses.py +++ /dev/null @@ -1,236 +0,0 @@ -""" -Cardano HD address derivation service. - -Derives Cardano base addresses from an account-level extended public key (xpub) -exported from wallets such as Eternl or Lace. Uses BIP-44 derivation via pycardano. - -Key derivation path: m / 1852' / 1815' / account' / chain / index - - chain 0 = external (receive) addresses - - chain 2 = staking key (always index 0 for the account) - -The xpub accepted here is the *account* public key — the root has already been -hardened away by the wallet. We only perform soft derivation from account level -down, so no private key material is ever needed or touched. -""" -import logging -from typing import Optional - -logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Public API -# --------------------------------------------------------------------------- - - -def derive_address(xpub_hex: str, index: int, network: str = "mainnet") -> str: - """ - Derive a Cardano base address at the given receive-address index. - - The address is a Shelley-era base address combining: - - payment key: account_xpub / 0 (external chain) / index - - staking key: account_xpub / 2 (staking chain) / 0 - - Args: - xpub_hex: Hex-encoded account extended public key (64 bytes raw or - 96 bytes with chain code, as exported by most CIP-1852 wallets). - index: Receive address index (0-based). Must be >= 0. - network: "mainnet" or "testnet" (preprod / preview). Defaults to mainnet. - - Returns: - Bech32-encoded Cardano base address (addr1... or addr_test1...). - - Raises: - ValueError: If xpub_hex is malformed, index is negative, or network is invalid. - RuntimeError: If pycardano is not installed or derivation fails unexpectedly. - """ - _require_pycardano() - - if index < 0: - raise ValueError(f"Address index must be non-negative, got {index}") - - net = _parse_network(network) - acct_pub = _parse_xpub(xpub_hex) - - try: - # External receive chain (0) / address index — soft (non-hardened) derivation. - addr_node = acct_pub.derive(0, private=False).derive(index, private=False) - # Staking chain (2) / always index 0 for the account. - stake_node = acct_pub.derive(2, private=False).derive(0, private=False) - except Exception as exc: - logger.exception("[cardano] Key derivation failed at index %d", index) - raise RuntimeError(f"Key derivation failed: {exc}") from exc - - from pycardano import ( - Address, - PaymentVerificationKey, - StakeVerificationKey, - ) - - pay_vk = PaymentVerificationKey.from_primitive(addr_node.public_key) - stake_vk = StakeVerificationKey.from_primitive(stake_node.public_key) - - address = Address( - payment_part=pay_vk.hash(), - staking_part=stake_vk.hash(), - network=net, - ) - - return str(address) - - -def validate_xpub(xpub_hex: str) -> bool: - """ - Validate that an xpub string is well-formed and parseable. - - Checks: - - Is a non-empty string - - Is valid hex - - Is a valid pycardano HDPublicKey (correct byte length, valid point on curve) - - Args: - xpub_hex: Hex-encoded account extended public key. - - Returns: - True if the xpub is valid, False otherwise. Never raises. - """ - if not xpub_hex or not isinstance(xpub_hex, str): - return False - - # Quick hex sanity before paying the crypto cost - stripped = xpub_hex.strip() - if not _is_hex(stripped): - return False - - try: - _require_pycardano() - node = _parse_xpub(stripped) - # Soft-derive a single child to prove the key is usable — HDWallet - # construction is lazy, so we actually exercise the BIP32 math. - node.derive(0, private=False) - return True - except Exception: - return False - - -def get_address_preview(xpub_hex: str, network: str = "mainnet") -> str: - """ - Derive the address at index 0 for settings UI preview. - - Thin wrapper around derive_address — exists so callers don't have to - know or care about the index convention. - - Args: - xpub_hex: Hex-encoded account extended public key. - network: "mainnet" or "testnet". Defaults to mainnet. - - Returns: - Bech32-encoded Cardano base address at index 0. - - Raises: - ValueError: If xpub_hex is malformed or network is invalid. - RuntimeError: If derivation fails unexpectedly. - """ - return derive_address(xpub_hex, index=0, network=network) - - -# --------------------------------------------------------------------------- -# Internal helpers -# --------------------------------------------------------------------------- - - -def _require_pycardano() -> None: - """Raise a clear RuntimeError if pycardano is not installed.""" - try: - import pycardano # noqa: F401 - except ImportError as exc: - raise RuntimeError( - "pycardano is required for Cardano address derivation. " - "Add pycardano>=0.11.0 to requirements.txt and reinstall." - ) from exc - - -def _parse_network(network: str): - """ - Parse a network string into a pycardano Network enum value. - - Args: - network: "mainnet" or "testnet". - - Returns: - pycardano.Network enum member. - - Raises: - ValueError: If network is not one of the accepted values. - """ - from pycardano import Network - - if network == "mainnet": - return Network.MAINNET - if network == "testnet": - return Network.TESTNET - raise ValueError( - f"Invalid network '{network}'. Expected 'mainnet' or 'testnet'." - ) - - -def _parse_xpub(xpub_hex: str): - """ - Parse a hex-encoded extended public key into a public-only HDWallet node. - - pycardano exposes soft-derivation through :class:`pycardano.HDWallet`. - An account-level xpub is 64 bytes (32-byte Ed25519 public key + - 32-byte chain code). Some wallets export 96 bytes; if so, we strip - the first 32 bytes which are typically a zeroed / duplicated prefix. - - Args: - xpub_hex: Hex-encoded extended public key string. - - Returns: - pycardano.HDWallet node rooted at the account level, with private - key fields unset. ``node.derive(index, private=False)`` performs - the soft CIP-1852 derivation we need. - - Raises: - ValueError: If the byte length is unexpected or the key is invalid. - """ - from pycardano import HDWallet - - try: - raw = bytes.fromhex(xpub_hex.strip()) - except ValueError as exc: - raise ValueError(f"xpub_hex is not valid hex: {exc}") from exc - - # Standard CIP-1852 account xpub is 64 bytes (pubkey || chain_code). - # Some export formats prepend 32 zeroed or duplicated bytes — handle both. - if len(raw) == 64: - pass # Expected format. - elif len(raw) == 96: - raw = raw[32:] - else: - raise ValueError( - f"Unexpected xpub length: {len(raw)} bytes. " - "Expected 64 bytes (pubkey + chain_code)." - ) - - public_key = raw[:32] - chain_code = raw[32:] - - try: - return HDWallet( - public_key=public_key, - chain_code=chain_code, - path="m/1852'/1815'/0'", - ) - except Exception as exc: - raise ValueError(f"xpub is not a valid extended public key: {exc}") from exc - - -def _is_hex(value: str) -> bool: - """Return True if every character in value is a valid hex digit.""" - if not value: - return False - try: - bytes.fromhex(value) - return True - except ValueError: - return False diff --git a/cardano_checkout/ipfs.py b/cardano_checkout/ipfs.py deleted file mode 100644 index 79a5e31..0000000 --- a/cardano_checkout/ipfs.py +++ /dev/null @@ -1,107 +0,0 @@ -"""Minimal IPFS client — upload + pin via kubo's HTTP API. - -Designed for the ``chromaticcraft`` shape: a small local kubo daemon -runs alongside the web app, accepts uploads from end users (e.g. Abby -uploading a photo of a finished custom order), pins locally for fast -serving, and optionally mirrors pins to a second remote node -(Lucy-on-LAN) for archival redundancy. - -No IPFS libraries are imported — just httpx against the kubo REST API -(v0). Keeps the SDK surface minimal. -""" - -from __future__ import annotations - -import logging -from dataclasses import dataclass -from typing import Optional - -import httpx - -logger = logging.getLogger(__name__) - - -@dataclass -class IPFSClient: - """Kubo-compatible IPFS client. - - Attributes: - api_url: Base URL of the kubo HTTP API (default ``http://127.0.0.1:5001``). - timeout: Per-request timeout in seconds (default 60 — uploads can be slow). - mirror_api_urls: Optional list of additional kubo endpoints to - ``pin add`` the CID on after a successful primary pin. Use this - to mirror to Lucy or any other archival node. - """ - - api_url: str = "http://127.0.0.1:5001" - timeout: float = 60.0 - mirror_api_urls: list[str] = None # type: ignore[assignment] - - def __post_init__(self) -> None: - if self.mirror_api_urls is None: - self.mirror_api_urls = [] - - async def add(self, data: bytes, filename: str = "upload") -> str: - """Upload bytes and pin them locally. - - Args: - data: Raw bytes to add. - filename: Logical name used by clients browsing the DAG - (doesn't affect the CID). - - Returns: - CID (base58, v0 or base32 v1 depending on kubo defaults). - - Raises: - RuntimeError: If the daemon is unreachable or returns a non-2xx. - """ - url = f"{self.api_url.rstrip('/')}/api/v0/add" - async with httpx.AsyncClient(timeout=self.timeout) as client: - resp = await client.post( - url, - files={"file": (filename, data, "application/octet-stream")}, - params={"pin": "true", "cid-version": "1"}, - ) - if resp.status_code >= 400: - raise RuntimeError(f"ipfs add {resp.status_code}: {resp.text[:200]}") - # kubo's /add streams NDJSON; each line is one {Name, Hash, Size}. - # For a single file upload the last line carries the wrapping CID. - last_cid: Optional[str] = None - for line in resp.text.strip().splitlines(): - if '"Hash"' in line: - import json - obj = json.loads(line) - last_cid = obj.get("Hash") - if not last_cid: - raise RuntimeError(f"ipfs add: no CID in response: {resp.text[:200]}") - - # Mirror pins (best effort — a mirror failure should not poison the primary upload). - for mirror in self.mirror_api_urls: - try: - await self._pin_on(mirror, last_cid) - except Exception as exc: - logger.warning("[ipfs] mirror pin to %s failed for %s: %s", mirror, last_cid, exc) - - return last_cid - - async def _pin_on(self, api_url: str, cid: str) -> None: - """Pin an existing CID on a remote kubo node.""" - url = f"{api_url.rstrip('/')}/api/v0/pin/add" - async with httpx.AsyncClient(timeout=self.timeout) as client: - resp = await client.post(url, params={"arg": cid}) - if resp.status_code >= 400: - raise RuntimeError(f"pin/add {resp.status_code}: {resp.text[:200]}") - - -async def pin_bytes( - data: bytes, - api_url: str = "http://127.0.0.1:5001", - mirror_api_urls: Optional[list[str]] = None, - filename: str = "upload", -) -> str: - """Convenience wrapper: one-shot upload + pin (+ optional mirror). - - Returns the CID. - """ - client = IPFSClient(api_url=api_url, mirror_api_urls=mirror_api_urls or []) - return await client.add(data, filename=filename) diff --git a/cardano_checkout/mint.py b/cardano_checkout/mint.py deleted file mode 100644 index cce81c3..0000000 --- a/cardano_checkout/mint.py +++ /dev/null @@ -1,442 +0,0 @@ -"""CIP-25 v2 NFT certificate-of-authenticity minting. - -This module produces the NFT cert attached to a confirmed merchant -order. One NFT per order, pinned-once metadata (image CID from IPFS -via :mod:`cardano_checkout.ipfs`), sent directly to the customer's -wallet in the same transaction. - -Design decisions: - -- **CIP-25 v2** (not CIP-68). CIP-25 is universally supported by - every Cardano wallet (Eternl, Lace, Yoroi, Vespr, Typhon). CIP-68 - adds reference-NFT mutability we do not need for a static cert. -- **Single policy per merchant studio.** All of a studio's certs share - one policy_id so wallets group them cleanly. The policy key is a - native script under the studio's custody — Sulkta pattern is a - multi-sig native script stored on Lucy. -- **Policy has a time-lock** (invalid-after slot) so the "no more - editions can be minted after X" claim is cryptographically enforceable. - Recommended: generous lock (100 years) so policy_id stays stable, - but revokable in-contract via ``mint policy revoke`` flow. -- **No reference script, no Plutus.** Pure native scripts + standard - CIP-25 metadata keeps the tx cheap (~0.18 ADA fee + min-utxo for the - NFT output). - -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 TYPE_CHECKING, Optional - -logger = logging.getLogger(__name__) - -if TYPE_CHECKING: # pragma: no cover — hints only - from pycardano import ChainContext - - -# --------------------------------------------------------------------------- -# Policy model -# --------------------------------------------------------------------------- - - -@dataclass -class MintPolicy: - """A native-script minting policy under the SDK's custody model. - - Attributes: - policy_id: Hex blake2b-224 hash of the native script CBOR. Stable - for the life of the policy — shipped with every cert minted - 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. - 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 - mathematically verifiable). - """ - - policy_id: str - script_cbor_hex: str - required_signer_hashes: list[str] = field(default_factory=list) - locked_after_slot: Optional[int] = None - - -@dataclass -class UnsignedMint: - """An unsigned mint transaction, ready to be handed to a cold signer. - - 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. - """ - - 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( - policy_id: str, - asset_name: str, - name: str, - image_cid: str, - description: str = "", - media_type: str = "image/jpeg", - properties: Optional[dict] = None, -) -> dict: - """Assemble the ``{721: {...}}`` metadatum envelope for a single NFT. - - CIP-25 v2 image field takes an ``ipfs://`` URI. Description, if - longer than 64 characters, is split into an array of ≤64-char chunks - (CIP-25 constraint from the Cardano metadata schema — strings larger - than 64 chars are encoded as a list of chunks). - - Args: - policy_id: Hex policy id (same as on the asset). - asset_name: UTF-8 asset name — used as the dict key under policy_id. - name: Human-readable NFT title (shown in wallets). - image_cid: IPFS CID — the function prepends ``ipfs://``. - description: Optional longer text. Will be chunked if > 64 chars. - media_type: MIME type of the image. Default ``image/jpeg``. - properties: Additional key/value pairs merged into the metadata blob. - - Returns: - Dict ready to submit as tx metadatum label 721. - """ - - def chunk64(s: str) -> list[str]: - if len(s) <= 64: - return [s] - return [s[i : i + 64] for i in range(0, len(s), 64)] - - desc: object = description - if isinstance(description, str) and len(description) > 64: - desc = chunk64(description) - - body: dict = { - "name": name, - "image": f"ipfs://{image_cid}", - "mediaType": media_type, - } - if desc: - body["description"] = desc - if properties: - body.update(properties) - - return { - "721": { - policy_id: { - asset_name: body, - }, - "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 diff --git a/cardano_checkout/monitor.py b/cardano_checkout/monitor.py index e79f4ba..bffb8c4 100644 --- a/cardano_checkout/monitor.py +++ b/cardano_checkout/monitor.py @@ -30,19 +30,23 @@ from __future__ import annotations import logging from datetime import datetime, timedelta, timezone -from typing import Optional +from typing import Awaitable, Callable, Optional import httpx from cardano_checkout.invoice import Invoice, InvoiceStatus -from cardano_checkout.oracles import ( - KNOWN_TOKENS, - convert_token_to_lovelace, - convert_usd_to_lovelace, - get_ada_usd_price, -) from cardano_checkout.store import InvoiceStore +# Consumer-supplied pricing callable: takes a USD amount (float), +# returns the current-market lovelace equivalent as int. Invoked by +# :func:`reprice_expired_invoices` to generate fresh quotes when an +# invoice's quote window lapses without payment. +# +# SDK intentionally does NOT ship an oracle. Consumers wire whatever +# price source they trust (CoinGecko, Koios ticker, their own DEX feed, +# or a constant for tests). +PriceFn = Callable[[float], Awaitable[int]] + logger = logging.getLogger(__name__) KOIOS_URL = "https://api.koios.rest/api/v1/address_utxos" @@ -164,33 +168,12 @@ async def evaluate_utxos( if qty > 0: received_assets[asset_id] = received_assets.get(asset_id, 0) + qty - # Convert native assets to lovelace equivalent via DexHunter. - asset_lovelace = 0 - for asset_id, qty in received_assets.items(): - if "." not in asset_id: - continue - policy_id, asset_name_hex = asset_id.split(".", 1) - - decimals = 0 - for token_info in KNOWN_TOKENS.values(): - if token_info.get("policy_id") == policy_id: - decimals = token_info.get("decimals", 0) - break - - try: - lv = await convert_token_to_lovelace( - policy_id, asset_name_hex, qty, decimals - ) - if lv is not None: - asset_lovelace += lv - except Exception as e: - logger.warning( - "[cardano-monitor] Failed to convert asset %s to lovelace: %s", - asset_id[:20], - e, - ) - - total_value = raw_lovelace + asset_lovelace + # ADA-only matching. Any native tokens landed in the same UTxOs are + # recorded in received_assets for visibility but do NOT contribute to + # the payment-matched total. Consumers who want to accept stablecoins + # or other native tokens wrap this function with their own asset-to- + # lovelace converter before comparing against expected_lovelace. + total_value = raw_lovelace if expected_lovelace == 0: # Degenerate case — any payment at all counts. @@ -313,23 +296,35 @@ async def check_pending_invoices( async def reprice_expired_invoices( store: InvoiceStore, + *, + price_fn: PriceFn, window_minutes: int = DEFAULT_PAYMENT_WINDOW_MINUTES, max_repricings: int = DEFAULT_MAX_REPRICINGS, limit: int = 100, ) -> int: """Reprice PENDING invoices whose expiry has passed. - Pulls the current ADA/USD oracle price, recalculates ``expected_lovelace`` - from the invoice's ``usd_amount``, resets ``expires_at`` to - ``now + window_minutes``, and tracks reprice count in ``invoice.metadata`` - under the key ``repriced_count``. After ``max_repricings`` the invoice - is transitioned to :class:`InvoiceStatus.EXPIRED`. + Calls the consumer-supplied ``price_fn(usd_amount) -> lovelace`` to + recompute ``expected_lovelace`` at current market. Resets ``expires_at`` + to ``now + window_minutes`` and increments ``invoice.metadata["repriced_count"]``. + After ``max_repricings`` the invoice transitions to + :class:`InvoiceStatus.EXPIRED`. Args: store: Persistence backend. - window_minutes: New expiry window per reprice. Matches TradeCraft's - platform-config-driven value of 15 minutes by default. - max_repricings: Give-up threshold. TradeCraft default is 3. + price_fn: Async callable that takes a USD amount and returns the + current lovelace equivalent. Consumer-supplied — the SDK does + not ship an oracle. A simple wiring looks like:: + + from cardano_checkout.monitor import reprice_expired_invoices + + async def my_price_fn(usd: float) -> int: + rate = await coingecko_fetch_ada_usd() # your code + return int(round(usd / rate * 1_000_000)) + + await reprice_expired_invoices(store, price_fn=my_price_fn) + window_minutes: New expiry window per reprice. TradeCraft default 15. + max_repricings: Give-up threshold. TradeCraft default 3. limit: Max pending invoices to process per call. Returns: @@ -350,13 +345,6 @@ async def reprice_expired_invoices( "[cardano-monitor] Repricing %d expired invoice(s)", len(expired_candidates) ) - ada_price = await get_ada_usd_price() - if ada_price <= 0: - logger.warning( - "[cardano-monitor] Cannot reprice — ADA price unavailable" - ) - return 0 - new_expires_at = now + timedelta(minutes=window_minutes) updated = 0 @@ -386,11 +374,20 @@ async def reprice_expired_invoices( ) continue - new_lovelace = await convert_usd_to_lovelace(usd_amount) - if new_lovelace == 0: + try: + new_lovelace = await price_fn(usd_amount) + except Exception as e: logger.warning( - "[cardano-monitor] invoice %s: lovelace conversion returned 0, skipping", + "[cardano-monitor] invoice %s: price_fn raised %s, skipping", invoice.id, + e, + ) + continue + if new_lovelace <= 0: + logger.warning( + "[cardano-monitor] invoice %s: price_fn returned %d, skipping", + invoice.id, + new_lovelace, ) continue @@ -398,18 +395,15 @@ async def reprice_expired_invoices( invoice.expected_lovelace = new_lovelace invoice.expires_at = new_expires_at invoice.metadata["repriced_count"] = repriced_count + 1 - invoice.metadata["ada_price_usd"] = round(ada_price, 4) await store.update(invoice) updated += 1 logger.info( - "[cardano-monitor] Repriced invoice %s: %d -> %d lovelace " - "(ADA=$%.4f, reprice #%d)", + "[cardano-monitor] Repriced invoice %s: %d -> %d lovelace (reprice #%d)", invoice.id, old_lovelace or 0, new_lovelace, - ada_price, repriced_count + 1, ) diff --git a/cardano_checkout/oracles.py b/cardano_checkout/oracles.py deleted file mode 100644 index ec64b7a..0000000 --- a/cardano_checkout/oracles.py +++ /dev/null @@ -1,346 +0,0 @@ -""" -Cardano Token Price Service — Phase 2 of the Cardano payments system. - -Provides cached ADA/USD and token/ADA price lookups used to convert -invoice amounts into lovelace (ADA's base unit) for payment requests. - -Data sources: - - ADA/USD: CoinGecko free API (no key required, rate-limited) - - Token/ADA: DexHunter v2 API (DEX aggregator on Cardano) - -Cache strategy: module-level dict with timestamps. TTL = 5 minutes. -All functions are async, never raise — return None/0 on failure. -""" - -import logging -import time -from typing import Optional - -import httpx - -logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Token registry -# --------------------------------------------------------------------------- - -KNOWN_TOKENS: dict[str, dict] = { - "ada": { - "policy_id": "", - "asset_name": "", - "ticker": "ADA", - "decimals": 6, - "type": "native", - }, - "djed": { - "policy_id": "8db269c3ec630e06ae29f74bc39edd1f87c819f1056206e879a1cd61", - "asset_name": "444a4544", # "DJED".encode().hex() - "ticker": "DJED", - "decimals": 6, - "type": "stablecoin", - }, - "iusd": { - "policy_id": "f66d78b4a3cb3d37afa0ec36461e51ecbde00f26c8f0a68f94b69880", - "asset_name": "69555344", # "iUSD".encode().hex() - "ticker": "iUSD", - "decimals": 6, - "type": "stablecoin", - }, - "night": { - "policy_id": "0691b2fecca1ac4f53cb6dfb00b7013e561d1f34403b957cbb5af1fa", - "asset_name": "4e49474854", # "NIGHT".encode().hex() - "ticker": "NIGHT", - "decimals": 6, - "type": "utility", - }, - "snek": { - "policy_id": "279c909f348e533da5808898f87f9a14bb2c3dfbbacccd631d927a3f", - "asset_name": "534e454b", # "SNEK".encode().hex() - "ticker": "SNEK", - "decimals": 0, - "type": "meme", - }, - "iag": { - "policy_id": "5d16944c1e00a5fa1d14ba2460709bc2e41a18e8e1b86a1e7a09da09", - "asset_name": "494147", # "IAG".encode().hex() - "ticker": "IAG", - "decimals": 6, - "type": "utility", - }, -} - -# --------------------------------------------------------------------------- -# Internal cache — { key: (value, fetched_at_unix) } -# --------------------------------------------------------------------------- - -_CACHE: dict[str, tuple] = {} -_CACHE_TTL_SECONDS = 300 # 5 minutes - - -def _cache_get(key: str) -> Optional[float]: - """Return cached value if still fresh, else None.""" - entry = _CACHE.get(key) - if entry is None: - return None - value, fetched_at = entry - if time.monotonic() - fetched_at > _CACHE_TTL_SECONDS: - return None - return value - - -def _cache_set(key: str, value: float) -> None: - """Store value in cache with current timestamp.""" - _CACHE[key] = (value, time.monotonic()) - - -# --------------------------------------------------------------------------- -# Public API -# --------------------------------------------------------------------------- - - -async def get_ada_usd_price() -> float: - """ - Fetch the current ADA/USD price from CoinGecko. - - Caches result for 5 minutes. Returns 0.0 on failure — callers should - treat 0.0 as a signal that pricing is unavailable. - - Endpoint: GET https://api.coingecko.com/api/v3/simple/price - """ - cache_key = "ada_usd" - cached = _cache_get(cache_key) - if cached is not None: - return cached - - url = "https://api.coingecko.com/api/v3/simple/price" - params = {"ids": "cardano", "vs_currencies": "usd"} - - try: - async with httpx.AsyncClient(timeout=10) as client: - resp = await client.get(url, params=params) - resp.raise_for_status() - data = resp.json() - price = float(data["cardano"]["usd"]) - - except httpx.HTTPStatusError as e: - logger.error( - "[cardano_price] CoinGecko request failed: %s %s", - e.response.status_code, - e.response.text[:200], - ) - return 0.0 - except (KeyError, ValueError, TypeError) as e: - logger.error("[cardano_price] CoinGecko response parse error: %s", e) - return 0.0 - except Exception as e: - logger.error("[cardano_price] CoinGecko unexpected error: %s", e) - return 0.0 - - logger.debug("[cardano_price] ADA/USD = %.6f (live)", price) - _cache_set(cache_key, price) - return price - - -async def get_token_ada_price(policy_id: str, asset_name_hex: str) -> Optional[float]: - """ - Fetch the price of a Cardano native token in ADA from DexHunter. - - Tries the DexHunter v2 bestPool endpoint first, then falls back to the - community pair endpoint. Both return the token's ADA price per base unit. - - Args: - policy_id: The token's Cardano policy ID (hex string). - asset_name_hex: The token's asset name as a hex-encoded string. - Derive with: token_ticker.encode().hex() - - Returns: - Price in ADA per base unit of the token, or None if no liquidity / - not found / request failed. - - Cache: 5 minutes per (policy_id, asset_name_hex) pair. - """ - if not policy_id or asset_name_hex is None: - # ADA itself — price is 1 ADA by definition - return 1.0 - - asset_id = f"{policy_id}{asset_name_hex}" - cache_key = f"token_ada:{asset_id}" - cached = _cache_get(cache_key) - if cached is not None: - return cached - - price: Optional[float] = None - - # --- Attempt 1: DexHunter v2 bestPool --- - try: - url = "https://api-v2.dexhunter.io/swap/bestPool" - params = {"tokenA": "lovelace", "tokenB": asset_id} - async with httpx.AsyncClient(timeout=10) as client: - resp = await client.get(url, params=params) - resp.raise_for_status() - data = resp.json() - - # DexHunter returns price_a_per_b or price_b_per_a depending on direction. - # We want ADA per token — look for the field that represents that. - raw_price = ( - data.get("price_b_per_a") # token per lovelace inverse - or data.get("price_a_per_b") # ada per token - or data.get("price") - ) - if raw_price is not None: - candidate = float(raw_price) - # bestPool returns lovelace-denominated prices — convert to ADA - # If the value is very large (>1000), it's likely lovelace/token, invert & divide - if candidate > 1000: - price = 1_000_000 / candidate # lovelace per token → ADA per token - else: - price = candidate - logger.debug("[cardano_price] %s bestPool price = %.8f ADA", asset_id[:20], price) - - except httpx.HTTPStatusError as e: - if e.response.status_code not in (404, 422): - logger.warning( - "[cardano_price] DexHunter bestPool error %s for %s", - e.response.status_code, - asset_id[:20], - ) - except Exception as e: - logger.warning("[cardano_price] DexHunter bestPool failed for %s: %s", asset_id[:20], e) - - # --- Attempt 2: DexHunter community pair endpoint (fallback) --- - if price is None: - try: - url = f"https://api.dexhunter.io/community/pair/{asset_id}" - async with httpx.AsyncClient(timeout=10) as client: - resp = await client.get(url) - resp.raise_for_status() - data = resp.json() - - raw_price = ( - data.get("price_ada") - or data.get("priceAda") - or data.get("price") - ) - if raw_price is not None: - price = float(raw_price) - logger.debug( - "[cardano_price] %s community pair price = %.8f ADA", - asset_id[:20], - price, - ) - - except httpx.HTTPStatusError as e: - if e.response.status_code not in (404, 422): - logger.warning( - "[cardano_price] DexHunter community error %s for %s", - e.response.status_code, - asset_id[:20], - ) - except Exception as e: - logger.warning("[cardano_price] DexHunter community failed for %s: %s", asset_id[:20], e) - - if price is not None and price > 0: - _cache_set(cache_key, price) - return price - - logger.info("[cardano_price] No price found for %s (no liquidity or unsupported)", asset_id[:20]) - return None - - -async def convert_usd_to_lovelace(usd_amount: float) -> int: - """ - Convert a USD amount to lovelace using the current ADA/USD price. - - 1 ADA = 1,000,000 lovelace. - - Args: - usd_amount: Amount in USD (e.g. 49.99). - - Returns: - Equivalent lovelace as an integer, or 0 if ADA price is unavailable. - - Example: - >>> await convert_usd_to_lovelace(10.00) - # At ADA = $0.45 → 10 / 0.45 ADA → 22,222,222 lovelace - """ - if usd_amount <= 0: - return 0 - - ada_usd = await get_ada_usd_price() - if ada_usd <= 0: - logger.error("[cardano_price] Cannot convert USD to lovelace — ADA price unavailable") - return 0 - - ada_amount = usd_amount / ada_usd - lovelace = int(ada_amount * 1_000_000) - - logger.debug( - "[cardano_price] $%.2f USD → %.6f ADA → %d lovelace (rate: $%.6f/ADA)", - usd_amount, - ada_amount, - lovelace, - ada_usd, - ) - return lovelace - - -async def convert_token_to_lovelace( - policy_id: str, - asset_name_hex: str, - token_quantity: int, - token_decimals: int = 0, -) -> Optional[int]: - """ - Convert a raw token quantity to its equivalent lovelace value. - - Uses the token's ADA price from DexHunter and accounts for decimal - precision so that, for example, 1,000,000 units of a 6-decimal token - equals 1.0 whole token. - - Args: - policy_id: Token policy ID. - asset_name_hex: Token asset name as hex (e.g. "534e454b" for SNEK). - token_quantity: Raw on-chain token quantity (base units, not decimal-adjusted). - token_decimals: Number of decimal places for the token (default 0). - - Returns: - Equivalent lovelace as an integer, or None if price is unavailable. - - Example: - # NIGHT token at 0.001 ADA/NIGHT, 6 decimals - # quantity = 5_000_000 (= 5.0 NIGHT), price = 0.001 ADA/token - # → 5.0 * 0.001 ADA = 0.005 ADA = 5,000 lovelace - >>> await convert_token_to_lovelace(policy_id, asset_name_hex, 5_000_000, 6) - 5000 - """ - if token_quantity <= 0: - return 0 - - # ADA is always 1:1 with itself in lovelace terms - if not policy_id and not asset_name_hex: - return token_quantity # already in lovelace - - token_ada_price = await get_token_ada_price(policy_id, asset_name_hex) - if token_ada_price is None: - logger.warning( - "[cardano_price] Cannot convert token to lovelace — no price for %s%s", - policy_id[:12], - asset_name_hex[:8], - ) - return None - - # Adjust for decimals: base_units / 10^decimals = whole tokens - whole_tokens = token_quantity / (10 ** token_decimals) - - # Whole tokens × ADA per token × lovelace per ADA - lovelace = int(whole_tokens * token_ada_price * 1_000_000) - - logger.debug( - "[cardano_price] %d base units (decimals=%d) → %.6f tokens × %.8f ADA → %d lovelace", - token_quantity, - token_decimals, - whole_tokens, - token_ada_price, - lovelace, - ) - return lovelace diff --git a/cardano_checkout/scheduler.py b/cardano_checkout/scheduler.py index c53cec2..557d6d9 100644 --- a/cardano_checkout/scheduler.py +++ b/cardano_checkout/scheduler.py @@ -37,6 +37,7 @@ from cardano_checkout.monitor import ( DEFAULT_MAX_REPRICINGS, DEFAULT_PAYMENT_WINDOW_MINUTES, KOIOS_URL, + PriceFn, check_pending_invoices, reprice_expired_invoices, ) @@ -64,6 +65,7 @@ class InvoiceScheduler: """ store: InvoiceStore + price_fn: Optional[PriceFn] = None koios_url: str = KOIOS_URL check_interval_seconds: int = 15 reprice_interval_seconds: int = 60 @@ -84,9 +86,15 @@ class InvoiceScheduler: ) async def _job_reprice_expired(self) -> None: + if self.price_fn is None: + # No oracle wired — skip repricing silently. Consumers that + # don't care about the USD-lock workflow (e.g. fixed-ADA + # invoices) will never configure a price_fn; that's fine. + return try: await reprice_expired_invoices( self.store, + price_fn=self.price_fn, window_minutes=self.payment_window_minutes, max_repricings=self.max_repricings, limit=self.limit, diff --git a/cardano_checkout/tradecraft_compat.py b/cardano_checkout/tradecraft_compat.py deleted file mode 100644 index 23ecdc6..0000000 --- a/cardano_checkout/tradecraft_compat.py +++ /dev/null @@ -1,440 +0,0 @@ -"""TradeCraft-specific compatibility shim. - -TradeCraft's ``services/cardano_scheduler.py`` shipped five jobs, of which -only two — ``check_pending_payments`` and ``reprice_expired_payments`` — -are generic invoice logic. The remaining three -(``_check_subscription_payments``, ``_reprice_subscription_payments``, -``_enforce_grace_period``) manipulate TradeCraft's ``Company``, -``Subscription``, and ``SubscriptionPayment`` SQLAlchemy models directly; -those are merchant-specific concerns that do not belong in the generic SDK. - -This module preserves the original TradeCraft import surface so the -existing TradeCraft code path still works while the migration to -:class:`cardano_checkout.store.InvoiceStore` is in-flight. None of the -symbols here are meant to be used by new consumers. - -**Do not depend on this module outside TradeCraft.** It is scheduled to -be removed once TradeCraft migrates fully to the Protocol-based API (see -TODO in the repo README). - -All functions in here are *verbatim* lifts from the original -``services/cardano_scheduler.py`` — TradeCraft's ``models`` + ``database`` -modules are imported lazily so that importing this module never fails -for non-TradeCraft consumers. If TradeCraft's models are not importable -the jobs raise at call time, not at import time. -""" - -from __future__ import annotations - -import logging -from datetime import datetime, timedelta, timezone -from decimal import Decimal -from typing import Optional - -from apscheduler.schedulers.asyncio import AsyncIOScheduler -from apscheduler.triggers.cron import CronTrigger -from apscheduler.triggers.interval import IntervalTrigger - -from cardano_checkout.monitor import check_address_utxos, evaluate_utxos -from cardano_checkout.oracles import convert_usd_to_lovelace, get_ada_usd_price - -logger = logging.getLogger(__name__) - - -# Original TradeCraft free-function names — keep these stable. -_check_address_utxos = check_address_utxos -_evaluate_payment = evaluate_utxos - - -def _require_tradecraft_models(): - """Lazy import of the TradeCraft-specific models + session maker. - - Returns ``(Company, Subscription, SubscriptionPayment, async_session_maker)``. - Raises ImportError with a clear message if TradeCraft isn't installed. - """ - try: - from database import async_session_maker # type: ignore[import-not-found] - from models import ( # type: ignore[import-not-found] - Company, - Subscription, - SubscriptionPayment, - ) - except ImportError as exc: # pragma: no cover — only meaningful inside TradeCraft - raise ImportError( - "cardano_checkout.tradecraft_compat requires the TradeCraft app's " - "`models` + `database` modules to be importable. This shim is " - "only meant to be used from within TradeCraft itself." - ) from exc - return Company, Subscription, SubscriptionPayment, async_session_maker - - -# --------------------------------------------------------------------------- -# Subscription payments job — verbatim TradeCraft logic -# --------------------------------------------------------------------------- - - -async def check_subscription_payments() -> None: # pragma: no cover — TradeCraft-only - """Poll Koios for UTXOs at awaiting_payment subscription addresses. - - On confirmation, advances ``Subscription.status`` to ``"active"`` and - updates ``Company.subscription_tier``. TradeCraft-specific. - """ - from sqlalchemy import select - - Company, Subscription, SubscriptionPayment, async_session_maker = ( - _require_tradecraft_models() - ) - - try: - async with async_session_maker() as db: - now = datetime.now(timezone.utc) - - result = await db.execute( - select(SubscriptionPayment).where( - SubscriptionPayment.status.in_(["awaiting_payment", "underpaid"]), - SubscriptionPayment.expires_at > now, - ) - ) - payments = result.scalars().all() - - if not payments: - return - - logger.debug( - "[tradecraft-compat] Checking %d subscription payment(s)", - len(payments), - ) - - for sp in payments: - try: - utxos = await _check_address_utxos(sp.address) - expected = sp.expected_lovelace or 0 - ( - new_status_enum, - raw_lovelace, - total_value, - received_assets, - tx_hash, - ) = await _evaluate_payment(expected, utxos) - new_status = new_status_enum.value - - status_map = { - "pending": "awaiting_payment", - "confirmed": "confirmed", - "overpaid": "overpaid", - "underpaid": "underpaid", - } - mapped_status = status_map.get(new_status, new_status) - - if mapped_status == sp.status and raw_lovelace == 0: - continue - - sp.received_lovelace = raw_lovelace - sp.total_value_lovelace = total_value - sp.received_assets = received_assets - - if tx_hash: - sp.tx_hash = tx_hash - - if mapped_status != sp.status: - old_status = sp.status - sp.status = mapped_status - - if mapped_status in ("confirmed", "overpaid"): - sp.confirmed_at = now - - if sp.subscription_id: - sub_result = await db.execute( - select(Subscription).where( - Subscription.id == sp.subscription_id - ) - ) - sub = sub_result.scalar_one_or_none() - if sub: - sub.status = "active" - sub.updated_at = now - if ( - sub.pending_tier - and sp.period_end - and sp.period_end <= now.date() - ): - sub.tier = sub.pending_tier - sub.pending_tier = None - sub.pending_tier_at = None - - company_result = await db.execute( - select(Company).where(Company.id == sp.company_id) - ) - company = company_result.scalar_one_or_none() - if company: - sub_result2 = await db.execute( - select(Subscription).where( - Subscription.company_id == sp.company_id - ) - ) - sub2 = sub_result2.scalar_one_or_none() - if sub2: - company.subscription_tier = sub2.tier - company.subscription_status = "active" - - logger.info( - "[tradecraft-compat] sub_payment #%d company_id=%d: " - "%s -> %s (%.6f ADA received)", - sp.id, - sp.company_id, - old_status, - mapped_status, - raw_lovelace / 1_000_000, - ) - - except Exception as e: - logger.exception( - "[tradecraft-compat] Error checking sub_payment #%d: %s", - sp.id, - e, - ) - - await db.commit() - - except Exception: - logger.exception( - "[tradecraft-compat] check_subscription_payments job failed" - ) - - -async def reprice_subscription_payments() -> None: # pragma: no cover — TradeCraft-only - """Reprice expired subscription payments — 24h window, 3-reprice cap.""" - from sqlalchemy import select - - _, _, SubscriptionPayment, async_session_maker = _require_tradecraft_models() - - try: - async with async_session_maker() as db: - now = datetime.now(timezone.utc) - - result = await db.execute( - select(SubscriptionPayment).where( - SubscriptionPayment.status == "awaiting_payment", - SubscriptionPayment.expires_at <= now, - SubscriptionPayment.repriced_count < 3, - ) - ) - payments = result.scalars().all() - - if not payments: - return - - logger.info( - "[tradecraft-compat] Repricing %d subscription payment(s)", - len(payments), - ) - - ada_price = await get_ada_usd_price() - if ada_price <= 0: - logger.warning( - "[tradecraft-compat] Cannot reprice subscriptions — " - "ADA price unavailable" - ) - return - - new_expires_at = now + timedelta(hours=24) - - for sp in payments: - try: - total_usd = float(sp.expected_usd or 0) - if total_usd <= 0: - sp.status = "expired" - continue - - new_lovelace = await convert_usd_to_lovelace(total_usd) - if new_lovelace == 0: - continue - - old_lovelace = sp.expected_lovelace - sp.expected_lovelace = new_lovelace - sp.ada_price_usd = Decimal(str(round(ada_price, 4))) - sp.expires_at = new_expires_at - sp.repriced_count += 1 - - logger.info( - "[tradecraft-compat] Repriced sub_payment #%d: %d -> %d " - "lovelace (ADA=$%.4f, reprice #%d)", - sp.id, - old_lovelace or 0, - new_lovelace, - ada_price, - sp.repriced_count, - ) - - except Exception as e: - logger.exception( - "[tradecraft-compat] Error repricing sub_payment #%d: %s", - sp.id, - e, - ) - - expired_result = await db.execute( - select(SubscriptionPayment).where( - SubscriptionPayment.status == "awaiting_payment", - SubscriptionPayment.expires_at <= now, - SubscriptionPayment.repriced_count >= 3, - ) - ) - for sp in expired_result.scalars().all(): - sp.status = "expired" - logger.info( - "[tradecraft-compat] sub_payment #%d expired after %d repricings", - sp.id, - sp.repriced_count, - ) - - await db.commit() - - except Exception: - logger.exception( - "[tradecraft-compat] reprice_subscription_payments job failed" - ) - - -async def enforce_grace_period() -> None: # pragma: no cover — TradeCraft-only - """Daily grace-period enforcement — past_due / suspended transitions.""" - from sqlalchemy import select - - Company, Subscription, SubscriptionPayment, async_session_maker = ( - _require_tradecraft_models() - ) - - try: - async with async_session_maker() as db: - today = datetime.now(timezone.utc).date() - - overdue_result = await db.execute( - select(SubscriptionPayment).where( - SubscriptionPayment.status.in_( - ["awaiting_payment", "underpaid", "expired"] - ), - SubscriptionPayment.due_date < today, - ) - ) - overdue_payments = overdue_result.scalars().all() - - for sp in overdue_payments: - try: - sub_result = await db.execute( - select(Subscription).where( - Subscription.company_id == sp.company_id - ) - ) - sub = sub_result.scalar_one_or_none() - if not sub or sub.status in ("cancelled", "suspended"): - continue - - company_result = await db.execute( - select(Company).where(Company.id == sp.company_id) - ) - company = company_result.scalar_one_or_none() - - if sp.grace_deadline and today > sp.grace_deadline: - if sub.status != "suspended": - sub.status = "suspended" - sub.updated_at = datetime.now(timezone.utc) - if company: - company.subscription_status = "suspended" - logger.info( - "[tradecraft-compat] company_id=%d suspended " - "(grace deadline %s passed)", - sp.company_id, - sp.grace_deadline, - ) - elif sub.status == "active": - sub.status = "past_due" - sub.updated_at = datetime.now(timezone.utc) - if company: - company.subscription_status = "past_due" - logger.info( - "[tradecraft-compat] company_id=%d past_due " - "(due_date %s passed)", - sp.company_id, - sp.due_date, - ) - - except Exception as e: - logger.exception( - "[tradecraft-compat] Error enforcing grace period " - "for sub_payment #%d: %s", - sp.id, - e, - ) - - await db.commit() - - except Exception: - logger.exception( - "[tradecraft-compat] enforce_grace_period job failed" - ) - - -# --------------------------------------------------------------------------- -# Standalone TradeCraft scheduler — registers the subscription jobs only. -# --------------------------------------------------------------------------- - - -_tc_scheduler: Optional[AsyncIOScheduler] = None - - -async def start_tradecraft_scheduler() -> None: # pragma: no cover — TradeCraft-only - """Start ONLY the TradeCraft-specific subscription + grace-period jobs. - - The generic invoice jobs should be run via :class:`InvoiceScheduler` - against a ``SQLAlchemyInvoiceStore`` adapter (not shipped here — - TradeCraft is responsible for implementing it during the migration). - """ - global _tc_scheduler - - if _tc_scheduler and _tc_scheduler.running: - return - - _tc_scheduler = AsyncIOScheduler() - - _tc_scheduler.add_job( - check_subscription_payments, - trigger=IntervalTrigger(seconds=60), - id="tradecraft_check_sub_payments", - name="TradeCraft: Check Subscription Payments", - replace_existing=True, - max_instances=1, - coalesce=True, - ) - - _tc_scheduler.add_job( - reprice_subscription_payments, - trigger=IntervalTrigger(hours=6), - id="tradecraft_reprice_sub_payments", - name="TradeCraft: Reprice Subscription Payments", - replace_existing=True, - max_instances=1, - coalesce=True, - ) - - _tc_scheduler.add_job( - enforce_grace_period, - trigger=CronTrigger(hour=6, minute=0, timezone="UTC"), - id="tradecraft_enforce_grace", - name="TradeCraft: Enforce Subscription Grace Periods", - replace_existing=True, - max_instances=1, - coalesce=True, - ) - - _tc_scheduler.start() - logger.info( - "[tradecraft-compat] Started — subscription + grace-period jobs only" - ) - - -async def stop_tradecraft_scheduler() -> None: # pragma: no cover — TradeCraft-only - """Stop the TradeCraft-specific scheduler.""" - global _tc_scheduler - if _tc_scheduler: - _tc_scheduler.shutdown(wait=False) - _tc_scheduler = None diff --git a/cardano_checkout/txbuild.py b/cardano_checkout/txbuild.py deleted file mode 100644 index 7c8b6b1..0000000 --- a/cardano_checkout/txbuild.py +++ /dev/null @@ -1,207 +0,0 @@ -"""Transaction construction helpers wrapping PyCardano. - -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. - -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 - -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 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 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 []) - - -# --------------------------------------------------------------------------- -# 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." - ) diff --git a/docs/minting-workflow.md b/docs/minting-workflow.md deleted file mode 100644 index 0dfccf7..0000000 --- a/docs/minting-workflow.md +++ /dev/null @@ -1,211 +0,0 @@ -# NFT Cert-of-Authenticity Minting Workflow - -This document is the operator runbook for minting a CIP-25 v2 NFT -certificate with `cardano-checkout`. It describes the hot/cold split, -what each host is responsible for, and the exact sequence of bytes that -move between them. - -## Architectural shape - -``` - hot host (Rackham) cold host (Lucy) - ────────────────────── ──────────────────────── - ┌────────────────────┐ ┌──────────────────────┐ - │ cardano-node │ │ policy skey files │ - │ (mainnet, n2n 6000)│ │ - Cobb.skey │ - │ ogmios 127:1337 │ │ - Kayos.skey │ - └──────────┬─────────┘ └──────────┬───────────┘ - │ │ - ┌──────────▼─────────┐ (1) body CBOR │ - │ cardano-checkout │ ──────────────────────► │ - │ mint_nft_cert() │ ▼ - │ returns UnsignedMint│ ┌──────────────────────┐ - └──────────┬─────────┘ │ cardano-cli / offline│ - │ │ signer: │ - │ (2) signed tx CBOR │ transaction witness │ - │ ◄───────────────────────────────┤ transaction assemble│ - ┌──────────▼─────────┐ └──────────────────────┘ - │ submit_signed_tx() │ - │ context.submit_tx │ - └──────────┬─────────┘ - │ - ▼ - on-chain confirmation -``` - -Two separable boundaries: - -1. **Chain ↔ cold.** The hot host (the one running `cardano-node` + - `ogmios`) has *no* policy signing keys. It can observe the chain, - build transaction bodies, and submit signed blobs — it cannot mint. -2. **Cold ↔ operator.** The cold host holds the policy skeys and - nothing else. No network access, no daemons. Its only job is: take - the unsigned body CBOR, produce a witness, hand the signed CBOR - back. - -## Step-by-step - -### 1. Build the unsigned tx on the hot host - -```python -from cardano_checkout import mint_nft_cert, build_cip25_metadata, MintPolicy - -policy = MintPolicy( - policy_id="", - script_cbor_hex="", - required_signer_hashes=[ - "", # 28 bytes, hex - "", - ], - locked_after_slot=None, # or a generous future slot if time-locked -) - -metadata = build_cip25_metadata( - policy_id=policy.policy_id, - asset_name="ChromaticCraftCert0042", - name="Chromatic Craft Cert #0042", - image_cid="bafybei...", # IPFS CID from IPFSClient.add() - description="Hand-stitched custom moth pendant", - media_type="image/png", - properties={ - "studio": "chromaticcraft", - "order_id": "CC-2026-0042", - "edition": "1 of 1", - }, -) - -unsigned = await mint_nft_cert( - policy=policy, - asset_name="ChromaticCraftCert0042", - metadata=metadata, - recipient_address="addr1q... (customer wallet)", - funding_address="addr1q... (chromaticcraft hot wallet)", - ogmios_host="127.0.0.1", - ogmios_port=1337, - network="mainnet", -) -``` - -Inspect `unsigned.summary` before sending the body anywhere — it's a -plaintext dump of what the mint is about to do. Operators should -eyeball the recipient address, the tx_id, the policy id, and the -required signers every single time. **This is the last chance to catch -a wrong asset name or recipient before the chain sees it.** - -Then materialise the three hex strings to disk: - -```bash -printf '%s' "$BODY_CBOR_HEX" > /tmp/mint-${TX_ID}.body.hex -printf '%s' "$AUX_CBOR_HEX" > /tmp/mint-${TX_ID}.aux.hex -printf '%s' "$SCRIPT_CBOR" > /tmp/mint-${TX_ID}.script.hex -``` - -Transfer those three files to the cold host. SCP over the wireguard -tunnel is fine. USB keys work too; QR codes work if the body is small -enough and the threat model demands strict air-gap. - -### 2. Sign on the cold host - -Reassemble a full transaction on cold, compute the witness, and emit -the signed CBOR. Using `cardano-cli`: - -```bash -# On the cold host, with Cobb.skey + Kayos.skey available. -cardano-cli transaction assemble \ - --tx-body-file /tmp/mint-${TX_ID}.body.json \ - --witness-file cobb.witness \ - --witness-file kayos.witness \ - --out-file /tmp/mint-${TX_ID}.signed.json -``` - -(Converting the `.hex` → `.json` form is `jq`-level — the SDK emits raw -CBOR because that's the smallest byte-shape to move around, but -`cardano-cli` expects the `{"type":"Witnessed Tx ConwayEra","cborHex":"..."}` -envelope.) - -A PyCardano-based offline signer can do the same thing without shelling -out: - -```python -from pycardano import ( - AuxiliaryData, NativeScript, PaymentExtendedSigningKey, - Transaction, TransactionBody, TransactionWitnessSet, - VerificationKeyWitness, -) - -body = TransactionBody.from_cbor(bytes.fromhex(body_hex)) -aux = AuxiliaryData.from_cbor(bytes.fromhex(aux_hex)) -script = NativeScript.from_cbor(bytes.fromhex(script_hex)) - -# Load skeys from disk — never log or print. -cobb_skey = PaymentExtendedSigningKey.load("./Cobb.skey") -kayos_skey = PaymentExtendedSigningKey.load("./Kayos.skey") - -# Each signer produces a vkey witness over the tx id. -witnesses = [ - VerificationKeyWitness( - vkey=skey.to_verification_key(), - signature=skey.sign(body.hash()), - ) - for skey in (cobb_skey, kayos_skey) -] - -tx = Transaction( - transaction_body=body, - transaction_witness_set=TransactionWitnessSet( - vkey_witnesses=witnesses, - native_scripts=[script], - ), - auxiliary_data=aux, -) - -signed_cbor_hex = tx.to_cbor_hex() -with open(f"/tmp/mint-{tx.id}.signed.hex", "w") as f: - f.write(signed_cbor_hex) -``` - -Confirm the tx id on cold matches the `unsigned.tx_id` the hot host -reported — the body is immutable so the two must match byte-for-byte. -If they don't, **stop**. Something rewrote the body between the two -hosts and signing it would broadcast a tx that doesn't match what was -intended. - -Transfer the signed CBOR back to the hot host. - -### 3. Submit from the hot host - -```python -from cardano_checkout import submit_signed_tx - -tx_hash = submit_signed_tx( - signed_tx_cbor_hex=signed_cbor_hex, - ogmios_host="127.0.0.1", - ogmios_port=1337, - network="mainnet", -) -# tx_hash matches unsigned.tx_id. Track it on cardanoscan or via Koios tx_info. -``` - -Ogmios returns immediately once the tx hits the mempool. Confirm on -chain by polling Koios `/tx_info` with the hash — typically lands in -the next 20-second block cycle. - -## Security notes - -- **Skey hygiene.** Skeys never leave the cold host. The hot host - never sees them. If you ever find yourself about to `scp` a skey, - stop and call it a day — there is no legitimate reason to move a - skey file. -- **Wipe temp files.** After submission, shred both the body and the - signed CBOR from any shared-storage mount. `shred -u`. -- **Time-locked policies.** If `policy.locked_after_slot` is set, the - SDK clamps the tx TTL to stay well inside the lock window. If the - chain tip has already passed the lock slot the node will reject the - witness — by design; that's what the time-lock buys you. -- **Ogmios is trusted.** The hot host trusts its local Ogmios to - surface the real chain tip. This is fine for a merchant mint path; - do *not* point `ogmios_host` at a third party you don't run. -- **Dry-run against preprod.** Every new policy gets smoke-tested on - `preprod` first. Build against a preprod xpub + preprod Ogmios, sign - on cold, submit. When you see the cert land in a preprod wallet, - flip over to mainnet. diff --git a/pyproject.toml b/pyproject.toml index 731b696..c3ab77a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,15 +4,15 @@ build-backend = "setuptools.build_meta" [project] name = "cardano-checkout" -version = "0.2.0.dev0" -description = "Merchant-side Cardano payments SDK + NFT cert-of-authenticity minting (zero-custody)" +version = "1.0.0.dev0" +description = "Merchant-side Cardano payment lifecycle (zero-custody). Ships the invoice + UTxO-watcher + reprice state machine. Use pycardano directly for Cardano primitives." readme = "README.md" requires-python = ">=3.10" license = {text = "Apache-2.0"} authors = [ {name = "Sulkta Coop"}, ] -keywords = ["cardano", "payments", "nft", "checkout", "blockchain", "ada", "pycardano"] +keywords = ["cardano", "payments", "checkout", "invoice", "utxo", "zero-custody", "merchant", "ada"] classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", @@ -26,7 +26,6 @@ classifiers = [ "Topic :: Software Development :: Libraries :: Python Modules", ] dependencies = [ - "pycardano>=0.11.0", "httpx>=0.27", "apscheduler>=3.10", ] diff --git a/tests/test_addresses.py b/tests/test_addresses.py deleted file mode 100644 index d060ee9..0000000 --- a/tests/test_addresses.py +++ /dev/null @@ -1,71 +0,0 @@ -"""Deterministic address-derivation smoke test. - -Uses a known test-vector xpub (the one shipped in the pycardano docs) -to assert the derived addresses are stable and reproducible across SDK -versions. If this test ever changes output, we have a backwards-compat -problem that would break every merchant's receive-address history. -""" - -from __future__ import annotations - -import pytest - -from cardano_checkout import addresses - - -# Public test vector — a CIP-1852 account extended public key. -# 64 bytes = 32 bytes Ed25519 pubkey || 32 bytes chain code, hex encoded. -# -# Derived deterministically from the well-known test mnemonic -# "test test test test test test test test test test test junk" -# at path m/1852'/1815'/0' via pycardano's HDWallet. Using a real, -# on-curve account xpub here (as opposed to random hex) is what lets -# validate_xpub + derive_address actually exercise the BIP32 math. -TEST_XPUB_HEX = ( - "f2cdeef60dfc2c00cd1d4c0def0ce3f7b0328f5badd2fd771f48ff207ca7eaa8" - "500a3c3d556f995e79c4a75e64d13ab12772f46e6c05fed1d9698b7e12a533f7" -) - - -def test_validate_xpub_accepts_well_formed_key() -> None: - assert addresses.validate_xpub(TEST_XPUB_HEX) is True - - -def test_validate_xpub_rejects_empty_and_junk() -> None: - assert addresses.validate_xpub("") is False - assert addresses.validate_xpub("notreallyhex!!") is False - assert addresses.validate_xpub("deadbeef") is False # wrong length - # Note: a correct-length random-hex string IS accepted — BIP32-ED25519 - # soft derivation over a 64-byte input doesn't require the public key - # half to be a point on the curve. We only catch shape errors here. - - -def test_derive_address_is_deterministic() -> None: - a0 = addresses.derive_address(TEST_XPUB_HEX, index=0, network="mainnet") - a0_again = addresses.derive_address(TEST_XPUB_HEX, index=0, network="mainnet") - assert a0 == a0_again - assert a0.startswith("addr1") - - -def test_derive_address_distinct_per_index() -> None: - a0 = addresses.derive_address(TEST_XPUB_HEX, index=0, network="mainnet") - a1 = addresses.derive_address(TEST_XPUB_HEX, index=1, network="mainnet") - a42 = addresses.derive_address(TEST_XPUB_HEX, index=42, network="mainnet") - assert a0 != a1 != a42 - - -def test_derive_address_network_switch_changes_prefix() -> None: - mainnet = addresses.derive_address(TEST_XPUB_HEX, index=0, network="mainnet") - testnet = addresses.derive_address(TEST_XPUB_HEX, index=0, network="testnet") - assert mainnet.startswith("addr1") - assert testnet.startswith("addr_test1") - - -def test_derive_address_rejects_negative_index() -> None: - with pytest.raises(ValueError, match="non-negative"): - addresses.derive_address(TEST_XPUB_HEX, index=-1) - - -def test_derive_address_rejects_bad_network() -> None: - with pytest.raises(ValueError, match="Invalid network"): - addresses.derive_address(TEST_XPUB_HEX, index=0, network="preprod") diff --git a/tests/test_cip25_metadata.py b/tests/test_cip25_metadata.py deleted file mode 100644 index ae495de..0000000 --- a/tests/test_cip25_metadata.py +++ /dev/null @@ -1,56 +0,0 @@ -"""CIP-25 v2 metadata envelope construction — pure unit tests, no network.""" - -from __future__ import annotations - -from cardano_checkout.mint import build_cip25_metadata - - -def test_basic_envelope_shape() -> None: - md = build_cip25_metadata( - policy_id="abc123", - asset_name="ChromaticCraft-Order-0001", - name="Chromatic Craft — Custom Order #0001", - image_cid="bafybeibgen", - description="Hand-stitched moth pendant", - properties={"order_id": "0001", "edition": "1 of 1"}, - ) - - assert md["721"]["version"] == "2.0" - assert "abc123" in md["721"] - - nft = md["721"]["abc123"]["ChromaticCraft-Order-0001"] - assert nft["name"] == "Chromatic Craft — Custom Order #0001" - assert nft["image"] == "ipfs://bafybeibgen" - assert nft["mediaType"] == "image/jpeg" - assert nft["description"] == "Hand-stitched moth pendant" - assert nft["order_id"] == "0001" - assert nft["edition"] == "1 of 1" - - -def test_description_under_64_chars_stays_a_string() -> None: - md = build_cip25_metadata( - policy_id="abc", asset_name="x", name="n", - image_cid="c", description="short", - ) - assert md["721"]["abc"]["x"]["description"] == "short" - - -def test_description_over_64_chars_chunks_to_list() -> None: - long = "x" * 150 - md = build_cip25_metadata( - policy_id="abc", asset_name="x", name="n", - image_cid="c", description=long, - ) - desc = md["721"]["abc"]["x"]["description"] - assert isinstance(desc, list) - assert all(len(chunk) <= 64 for chunk in desc) - assert "".join(desc) == long - - -def test_image_uri_has_ipfs_prefix() -> None: - md = build_cip25_metadata( - policy_id="abc", asset_name="x", name="n", - image_cid="bafybeitestcid", - ) - assert md["721"]["abc"]["x"]["image"].startswith("ipfs://") - assert "bafybeitestcid" in md["721"]["abc"]["x"]["image"] diff --git a/tests/test_mint_metadata.py b/tests/test_mint_metadata.py deleted file mode 100644 index ec4fb77..0000000 --- a/tests/test_mint_metadata.py +++ /dev/null @@ -1,308 +0,0 @@ -"""CIP-25 v2 envelope round-trips + unsigned-mint shape tests. - -Complements ``test_cip25_metadata.py`` (the pure builder unit tests) by -checking: - -- Round-tripping the envelope through pycardano's Metadata/AuxiliaryData - produces a well-formed CBOR blob (the wallet-visible thing). -- :func:`mint_nft_cert` returns a correctly-shaped :class:`UnsignedMint` - without hitting a live chain — we stub the ChainContext. - -The chain-context stub mirrors just enough of pycardano's interface for -``TransactionBuilder.build`` to succeed. No live Ogmios calls. -""" - -from __future__ import annotations - -import pytest - -from cardano_checkout import UnsignedMint, build_cip25_metadata, mint_nft_cert - - -# --------------------------------------------------------------------------- -# Fixtures — a deterministic test policy + a stub ChainContext -# --------------------------------------------------------------------------- - - -def _test_policy(): - """Return a fresh 2-of-2 NativeScript all-of policy + its hash + CBOR. - - Uses fixed verification-key hashes (32-char hex, 28 bytes as - required by Cardano VKH). No cryptographic significance — just - deterministic filler for tests. - """ - from pycardano import ( - NativeScript, - ScriptAll, - ScriptPubkey, - VerificationKeyHash, - ) - - vkh_cobb = VerificationKeyHash.from_primitive(bytes.fromhex("11" * 28)) - vkh_kayos = VerificationKeyHash.from_primitive(bytes.fromhex("22" * 28)) - script: NativeScript = ScriptAll( - [ScriptPubkey(vkh_cobb), ScriptPubkey(vkh_kayos)] - ) - return script, [vkh_cobb.payload.hex(), vkh_kayos.payload.hex()] - - -def _stub_context(): - """Return a minimal ChainContext stub with just enough surface for builder.build().""" - from pycardano import ( - Address, - AssetName, - MultiAsset, - ProtocolParameters, - TransactionId, - TransactionInput, - TransactionOutput, - UTxO, - Value, - ) - - class StubContext: - network = None - - @property - def last_block_slot(self) -> int: - return 100_000_000 - - @property - def protocol_param(self) -> ProtocolParameters: - # Mainnet values as of 2025-ish — just enough to get fee math through. - return ProtocolParameters( - min_fee_constant=155_381, - min_fee_coefficient=44, - max_block_size=90_112, - max_tx_size=16_384, - max_block_header_size=1_100, - key_deposit=2_000_000, - pool_deposit=500_000_000, - pool_influence=0.3, - monetary_expansion=0.003, - treasury_expansion=0.2, - decentralization_param=0, - extra_entropy="", - protocol_major_version=9, - protocol_minor_version=0, - min_utxo=1_000_000, - min_pool_cost=340_000_000, - price_mem=0.0577, - price_step=0.0000721, - max_tx_ex_mem=14_000_000, - max_tx_ex_steps=10_000_000_000, - max_block_ex_mem=62_000_000, - max_block_ex_steps=20_000_000_000, - max_val_size=5_000, - collateral_percent=150, - max_collateral_inputs=3, - coins_per_utxo_byte=4_310, - coins_per_utxo_word=34_482, - cost_models={}, # Plutus cost models — not used by native-script mints. - ) - - @property - def genesis_param(self): - # Minimal stand-in — pycardano's builder uses this for slot math. - from pycardano import GenesisParameters - - return GenesisParameters( - active_slots_coefficient=0.05, - update_quorum=5, - max_lovelace_supply=45_000_000_000_000_000, - network_magic=764_824_073, - epoch_length=432_000, - system_start=1_506_203_091, - slots_per_kes_period=129_600, - slot_length=1, - max_kes_evolutions=62, - security_param=2_160, - ) - - @property - def era(self): - from pycardano import Era - return Era.CONWAY - - def utxos(self, address): - # Provide one fat UTxO so the builder has inputs to draw fees + min-ADA from. - addr = ( - address - if isinstance(address, Address) - else Address.from_primitive(address) - ) - tx_in = TransactionInput( - transaction_id=TransactionId.from_primitive(bytes.fromhex("cc" * 32)), - index=0, - ) - # 1000 ADA, no native assets — plenty for fees + NFT min-UTxO. - output = TransactionOutput(addr, Value(1_000_000_000)) - return [UTxO(tx_in, output)] - - def submit_tx(self, tx): # pragma: no cover — not invoked in these tests - pass - - return StubContext() - - -# --------------------------------------------------------------------------- -# Test vectors for the envelope -# --------------------------------------------------------------------------- - - -def test_envelope_from_chromaticcraft_order_vector() -> None: - """A realistic chromaticcraft cert — image CID + studio properties.""" - md = build_cip25_metadata( - policy_id="4a3b2c1d0e9f8a7b6c5d4e3f2a1b0c9d8e7f6a5b4c3d2e1f0a9b8c7d", - asset_name="ChromaticCraftCert0042", - name="Chromatic Craft Cert #0042", - image_cid="bafybeihkop... real cid would be 59 base32 chars", - description=( - "Certificate of authenticity for hand-stitched custom moth pendant " - "ordered by Abby, completed 2026-04-17 by Cobb." - ), - media_type="image/png", - properties={ - "studio": "chromaticcraft", - "artisan": "Cobb", - "order_id": "CC-2026-0042", - "edition": "1 of 1", - "material": "sterling silver + polymer clay", - }, - ) - - label = md["721"] - nft = label[ - "4a3b2c1d0e9f8a7b6c5d4e3f2a1b0c9d8e7f6a5b4c3d2e1f0a9b8c7d" - ]["ChromaticCraftCert0042"] - - assert label["version"] == "2.0" - assert nft["name"] == "Chromatic Craft Cert #0042" - assert nft["mediaType"] == "image/png" - assert nft["studio"] == "chromaticcraft" - assert nft["artisan"] == "Cobb" - assert nft["edition"] == "1 of 1" - - -def test_envelope_roundtrips_through_pycardano_metadata() -> None: - """CBOR-encode and decode — the wallet-visible path.""" - from pycardano import AuxiliaryData, Metadata - - raw = build_cip25_metadata( - policy_id="ab" * 28, - asset_name="TestNFT", - name="Test NFT", - image_cid="bafybeitestcidforround-tripping", - description="round trip", - ) - # pycardano's Metadata expects int keys at the top level. - inner = {int(k): v for k, v in raw.items()} - md = Metadata(inner) - aux = AuxiliaryData(md) - - cbor_hex = aux.to_cbor_hex() - # Must be valid hex and non-trivially sized. - assert len(cbor_hex) > 40 - assert all(c in "0123456789abcdef" for c in cbor_hex) - - # Decoding back must yield the same metadata. - decoded = AuxiliaryData.from_cbor(bytes.fromhex(cbor_hex)) - decoded_md = decoded.data if hasattr(decoded, "data") else decoded # pycardano compat - assert decoded_md is not None - - -def test_envelope_version_key_always_present() -> None: - md = build_cip25_metadata( - policy_id="a" * 56, asset_name="x", name="n", image_cid="c" - ) - assert md["721"]["version"] == "2.0" - - -# --------------------------------------------------------------------------- -# mint_nft_cert — full tx body construction against a stubbed context -# --------------------------------------------------------------------------- - - -async def test_mint_nft_cert_returns_unsigned_bundle() -> None: - from pycardano import Address, Network, PaymentKeyPair, StakeKeyPair - - script, signer_hashes = _test_policy() - policy_cbor_hex = script.to_cbor_hex() - # policy_id is blake2b-224 of the script CBOR — use pycardano to compute. - policy_id = script.hash().payload.hex() - - # Build two throwaway addresses in testnet namespace. - pay_key = PaymentKeyPair.generate() - stk_key = StakeKeyPair.generate() - funding_addr = str( - Address( - payment_part=pay_key.verification_key.hash(), - staking_part=stk_key.verification_key.hash(), - network=Network.TESTNET, - ) - ) - recipient_pay = PaymentKeyPair.generate() - recipient_stk = StakeKeyPair.generate() - recipient_addr = str( - Address( - payment_part=recipient_pay.verification_key.hash(), - staking_part=recipient_stk.verification_key.hash(), - network=Network.TESTNET, - ) - ) - - from cardano_checkout import MintPolicy - - policy = MintPolicy( - policy_id=policy_id, - script_cbor_hex=policy_cbor_hex, - required_signer_hashes=signer_hashes, - ) - - metadata = build_cip25_metadata( - policy_id=policy_id, - asset_name="TestCert01", - name="Test Cert 01", - image_cid="bafybeitest", - ) - - result = await mint_nft_cert( - policy=policy, - asset_name="TestCert01", - metadata=metadata, - recipient_address=recipient_addr, - funding_address=funding_addr, - context=_stub_context(), - network="testnet", - ) - - assert isinstance(result, UnsignedMint) - assert len(result.tx_id) == 64 # hex-encoded 32-byte blake2b hash - assert result.tx_body_cbor_hex - assert all(c in "0123456789abcdef" for c in result.tx_body_cbor_hex) - assert result.auxiliary_data_cbor_hex - assert result.native_script_cbor_hex == policy_cbor_hex - assert result.required_signer_hashes == signer_hashes - assert policy_id in result.summary - assert recipient_addr in result.summary - - -async def test_mint_nft_cert_rejects_oversize_asset_name() -> None: - from cardano_checkout import MintPolicy - - policy = MintPolicy( - policy_id="aa" * 28, - script_cbor_hex="82008200581c" + "11" * 28, # doesn't matter — builder never reached - required_signer_hashes=[], - ) - - with pytest.raises(ValueError, match="asset_name"): - await mint_nft_cert( - policy=policy, - asset_name="X" * 33, # 33 > 32 byte limit - metadata={"721": {}}, - recipient_address="addr_test1...", - funding_address="addr_test1...", - context=_stub_context(), - network="testnet", - ) diff --git a/tests/test_monitor_with_inmemory_store.py b/tests/test_monitor_with_inmemory_store.py index 0c1507c..ae72ed2 100644 --- a/tests/test_monitor_with_inmemory_store.py +++ b/tests/test_monitor_with_inmemory_store.py @@ -53,23 +53,31 @@ def _utxo(lovelace: int, tx_hash: str = "aa" * 32) -> dict: @pytest.fixture(autouse=True) -def _patch_koios_and_oracle(monkeypatch): - """Default: no UTxOs, oracle returns $0.45/ADA. Individual tests override.""" +def _patch_koios(monkeypatch): + """Default: Koios returns no UTxOs. Individual tests override.""" async def fake_utxos(address, koios_url=None, timeout=None): return [] - async def fake_price(): - return 0.45 + monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos) - async def fake_convert(usd): + +@pytest.fixture +def price_fn_at_45c(): + """A deterministic price_fn for tests — USD priced at $0.45/ADA.""" + async def _convert(usd: float) -> int: if usd <= 0: return 0 return int((usd / 0.45) * 1_000_000) + return _convert - monkeypatch.setattr(monitor, "check_address_utxos", fake_utxos) - monkeypatch.setattr(monitor, "get_ada_usd_price", fake_price) - monkeypatch.setattr(monitor, "convert_usd_to_lovelace", fake_convert) + +@pytest.fixture +def price_fn_zero(): + """A price_fn that returns 0 — stand-in for oracle unavailability.""" + async def _zero(usd: float) -> int: + return 0 + return _zero # --------------------------------------------------------------------------- @@ -184,14 +192,14 @@ async def test_already_expired_invoices_are_skipped(monkeypatch) -> None: # --------------------------------------------------------------------------- -async def test_reprice_updates_expected_lovelace_and_extends_expiry() -> None: +async def test_reprice_updates_expected_lovelace_and_extends_expiry(price_fn_at_45c) -> None: store = InMemoryStore() inv = _make(expected_lovelace=5_000_000) inv.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) await store.create(inv) updated = await monitor.reprice_expired_invoices( - store, window_minutes=15, max_repricings=3 + store, price_fn=price_fn_at_45c, window_minutes=15, max_repricings=3 ) assert updated == 1 @@ -205,14 +213,14 @@ async def test_reprice_updates_expected_lovelace_and_extends_expiry() -> None: assert fetched.metadata["repriced_count"] == 1 -async def test_reprice_gives_up_after_max_repricings() -> None: +async def test_reprice_gives_up_after_max_repricings(price_fn_at_45c) -> None: store = InMemoryStore() inv = _make(expected_lovelace=5_000_000, repriced_count=3) inv.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) await store.create(inv) await monitor.reprice_expired_invoices( - store, window_minutes=15, max_repricings=3 + store, price_fn=price_fn_at_45c, window_minutes=15, max_repricings=3 ) fetched = await store.get("inv") @@ -220,26 +228,21 @@ async def test_reprice_gives_up_after_max_repricings() -> None: assert fetched.status == InvoiceStatus.EXPIRED -async def test_reprice_noop_when_nothing_expired() -> None: +async def test_reprice_noop_when_nothing_expired(price_fn_at_45c) -> None: store = InMemoryStore() - await store.create(_make(expired_in_minutes=15) if False else _make()) + await store.create(_make()) - updated = await monitor.reprice_expired_invoices(store) + updated = await monitor.reprice_expired_invoices(store, price_fn=price_fn_at_45c) assert updated == 0 -async def test_reprice_skips_when_oracle_unavailable(monkeypatch) -> None: +async def test_reprice_skips_when_oracle_returns_zero(price_fn_zero) -> None: store = InMemoryStore() inv = _make(expected_lovelace=5_000_000) inv.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1) await store.create(inv) - async def zero_price(): - return 0.0 - - monkeypatch.setattr(monitor, "get_ada_usd_price", zero_price) - - updated = await monitor.reprice_expired_invoices(store) + updated = await monitor.reprice_expired_invoices(store, price_fn=price_fn_zero) assert updated == 0 fetched = await store.get("inv")