"""Contract API.
Provides loading and interaction with CashScript-contract artifacts, including
constructor validation/encoding, address generation, and ABI-based unlockers.
"""
import os
from collections.abc import Callable
from typing import Any
import semver
from cashscript_py.helpers.address import script_to_address
from cashscript_py.helpers.argument_encoding import (
ConstructorArgument,
FunctionArgument,
encode_constructor_arguments,
encode_function_argument,
)
from cashscript_py.helpers.cashaddress import LockingBytecodeType, address_string_to_locking_bytecode
from cashscript_py.helpers.crypto import hash256
from cashscript_py.helpers.script import (
asm_to_script,
calculate_bytesize,
count_opcodes,
create_input_script,
generate_redeem_script,
serialize_script,
)
from cashscript_py.helpers.transaction import create_preimage
from cashscript_py.interfaces import ContractUnlocker, Output, Transaction, Utxo
from cashscript_py.network.network_provider import NetworkProvider
from cashscript_py.signature_template import SignatureTemplate
[docs]
class Contract:
"""High-level interface to a CashScript contract.
A Contract is constructed from a compiled cashc JSON artifact and constructor
arguments. It exposes ABI functions via the `unlock` mapping and provides
helper methods for querying UTXOs and balances.
Attributes:
name: The contract name from the artifact.
address: The contract address (token-unaware).
token_address: The contract address (token-aware).
bytecode: Contract redeem script bytecode (hex).
bytesize: Size of the redeem script in bytes.
opcount: Count of non-push opcodes in the redeem script.
provider: Network provider used for blockchain operations.
address_type: Locking bytecode type (e.g., P2SH32).
unlock: Mapping of function name to unlocker factory.
"""
def __init__(
self,
artifact: dict[str, Any],
constructor_args: list[ConstructorArgument],
provider: NetworkProvider,
address_type: LockingBytecodeType = LockingBytecodeType.P2SH32,
):
"""Initialize a Contract instance.
Args:
artifact: The compiled cashc JSON artifact.
constructor_args: Constructor arguments in ABI order.
provider: Network provider for queries and broadcasts.
address_type: Locking bytecode type to use for addresses (Default: P2SH32).
Raises:
ValueError: If the artifact is incomplete/invalid, the compiler
version is unsupported (< 0.7.0), or the constructor argument
count/types do not match the artifact.
"""
self.artifact = artifact
self.constructor_args: list[ConstructorArgument] = constructor_args
self.provider = provider
self.address_type = address_type
required_fields = ["abi", "bytecode", "constructorInputs", "contractName", "compiler"]
for field in required_fields:
if field not in artifact or artifact[field] is None:
raise ValueError(f"Missing {field}")
version = artifact["compiler"]["version"]
if not semver.VersionInfo.parse(version).match(">=0.7.0"):
raise ValueError(f"Artifact compiled with unsupported compiler version: {version}, required >=0.7.0")
if len(artifact["constructorInputs"]) != len(constructor_args):
expected_types = [input["type"] for input in artifact["constructorInputs"]]
raise ValueError(
f"Incorrect number of arguments passed to {artifact['contractName']} constructor. "
f"Expected {len(artifact['constructorInputs'])} arguments ({expected_types}) "
f"but got {len(constructor_args)}"
)
self.encoded_constructor_args: list[bytes] = encode_constructor_arguments(
constructor_args, artifact["constructorInputs"]
)
self.redeem_script = generate_redeem_script(asm_to_script(artifact["bytecode"]), self.encoded_constructor_args)
self.name = artifact["contractName"]
self.address = script_to_address(self.redeem_script, self.provider.network, self.address_type, False)
self.token_address = script_to_address(self.redeem_script, self.provider.network, self.address_type, True)
self.bytecode = serialize_script(self.redeem_script).hex()
self.bytesize = calculate_bytesize(self.redeem_script)
self.opcount = count_opcodes(self.redeem_script)
self.unlock: dict[str, Callable[..., ContractUnlocker]] = {}
if len(artifact["abi"]) == 1:
f = artifact["abi"][0]
self.unlock[f["name"]] = self._create_unlocker(f)
else:
for i, f in enumerate(artifact["abi"]):
self.unlock[f["name"]] = self._create_unlocker(f, i)
def _create_unlocker(
self, abi_function: dict[str, Any], selector: int | None = None
) -> Callable[..., ContractUnlocker]:
def unlocker(*args: FunctionArgument) -> ContractUnlocker:
if len(abi_function["inputs"]) != len(args):
expected_types = [inp["type"] for inp in abi_function["inputs"]]
raise ValueError(
f"Incorrect number of arguments passed to function {abi_function['name']}. "
f"Expected {len(abi_function['inputs'])} arguments ({expected_types}) "
f"but got {len(args)}"
)
encoded_args = [
encode_function_argument(arg, abi_function["inputs"][i]["type"]) for i, arg in enumerate(args)
]
def generate_unlocking_bytecode(
transaction: Transaction, source_outputs: list[Output], input_index: int
) -> bytes:
complete_args = []
for arg in encoded_args:
if isinstance(arg, SignatureTemplate):
# Build preimage and sighash (SIGHASH_UTXOS path handled by create_preimage)
hashtype_full = arg.get_hash_type().value
covered_script = serialize_script(self.redeem_script)
preimage = create_preimage(
transaction, source_outputs, input_index, covered_script, hashtype_full
)
sighash = hash256(preimage)
# Generate signature (sig + hashtype byte) for CHECKSIG
signature = arg.generate_signature(sighash)
# Debugging info
if os.environ.get("CASHSCRIPT_PY_DEBUG") == "1":
print(f"[contract-debug] function={abi_function.get('name')}")
print(
f"[contract-debug] hashtype_byte=0x{(hashtype_full & 0xFF):02x} "
f"algo={arg.get_signature_algorithm().name}"
)
print(f"[contract-debug] preimage: {preimage.hex()}")
print(f"[contract-debug] sighash: {sighash.hex()}")
print(f"[contract-debug] signature: {signature.hex()}")
complete_args.append(signature)
else:
complete_args.append(arg)
# Create input script
input_script = create_input_script(self.redeem_script, complete_args, selector)
# Debugging info
if os.environ.get("CASHSCRIPT_PY_DEBUG") == "1":
redeem_bc = serialize_script(self.redeem_script)
print(f"[contract-debug] redeem_script_bytecode: {redeem_bc.hex()}")
print(f"[contract-debug] input_script_bytecode: {input_script.hex()}")
return input_script
def generate_locking_bytecode() -> bytes:
return address_string_to_locking_bytecode(self.address)
return ContractUnlocker(
generate_locking_bytecode,
generate_unlocking_bytecode,
self,
list(args),
abi_function,
)
return unlocker
[docs]
async def get_balance(self) -> int:
"""Return the total balance (sum of UTXO satoshis) at the contract address.
Returns:
The sum of satoshis across all UTXOs at the contract address.
"""
utxos = await self.get_utxos()
return sum(utxo.satoshis for utxo in utxos)
[docs]
async def get_utxos(self) -> list[Utxo]:
"""Fetch all UTXOs (confirmed/unconfirmed) for the contract address.
Returns:
A list of UTXOs spendable by this contract.
"""
return await self.provider.get_utxos(self.address)