Merge branch 'main' into fix/minted_amount_requested

Author: cffls

integration-test/docker-compose.yml

@@ -52,7 +52,7 @@ services:
         max-file: "10"
 
   ogmios:
-    image: cardanosolutions/ogmios:latest
+    image: cardanosolutions/ogmios:v5.6.0-mainnet
     environment:
       NETWORK: "${NETWORK:-local-alonzo}"
 

integration-test/test/test_plutus.py

@@ -26,7 +26,7 @@ def test_plutus_v1(self):
 
         builder = TransactionBuilder(self.chain_context)
         builder.add_input_address(giver_address)
-        datum = PlutusData()  # A Unit type "()" in Haskell
+        datum = Unit()  # A Unit type "()" in Haskell
         builder.add_output(
             TransactionOutput(script_address, 50000000, datum_hash=datum_hash(datum))
         )
@@ -311,3 +311,63 @@ def test_plutus_v2_ref_script(self):
         self.chain_context.submit_tx(signed_tx)
 
         self.assert_output(taker_address, take_output)
+
+    @retry(tries=TEST_RETRIES, backoff=1.5, delay=6, jitter=(0, 4))
+    @pytest.mark.post_alonzo
+    def test_transaction_chaining(self):
+        giver_address = Address(self.payment_vkey.hash(), network=self.NETWORK)
+        builder = TransactionBuilder(self.chain_context)
+        builder.add_input_address(giver_address)
+        builder.add_output(TransactionOutput(giver_address, 50000000))
+        tx1 = builder.build_and_sign([self.payment_skey], giver_address)
+
+        utxo_to_spend = UTxO(
+            TransactionInput(tx1.id, 0), tx1.transaction_body.outputs[0]
+        )
+
+        builder = TransactionBuilder(self.chain_context)
+        builder.add_input(utxo_to_spend)
+        builder.add_output(TransactionOutput(giver_address, 25000000))
+        tx2 = builder.build_and_sign([self.payment_skey], giver_address)
+
+        self.chain_context.submit_tx(tx1)
+        self.chain_context.submit_tx(tx2)
+
+    @retry(tries=TEST_RETRIES, backoff=1.5, delay=6, jitter=(0, 4))
+    @pytest.mark.post_alonzo
+    def test_get_plutus_script(self):
+        # ----------- Giver give ---------------
+        with open("./plutus_scripts/fortytwoV2.plutus", "r") as f:
+            script_hex = f.read()
+            forty_two_script = PlutusV2Script(cbor2.loads(bytes.fromhex(script_hex)))
+
+        script_hash = plutus_script_hash(forty_two_script)
+
+        script_address = Address(script_hash, network=self.NETWORK)
+
+        giver_address = Address(self.payment_vkey.hash(), network=self.NETWORK)
+
+        builder = TransactionBuilder(self.chain_context)
+        builder.add_input_address(giver_address)
+        builder.add_output(
+            TransactionOutput(script_address, 50000000, script=forty_two_script)
+        )
+
+        signed_tx = builder.build_and_sign([self.payment_skey], giver_address)
+
+        print("############### Transaction created ###############")
+        print(signed_tx)
+        print(signed_tx.to_cbor_hex())
+        print("############### Submitting transaction ###############")
+        self.chain_context.submit_tx(signed_tx)
+        time.sleep(3)
+
+        utxos = self.chain_context.utxos(script_address)
+
+        assert utxos[0].output.script == forty_two_script
+
+
+class TestPlutusOgmiosOnly(TestPlutus):
+    @classmethod
+    def setup_class(cls):
+        cls.chain_context._kupo_url = None

pycardano/backend/ogmios.py

@@ -4,7 +4,6 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional, Tuple, Union
 
-import cbor2
 import requests
 import websocket
 from cachetools import Cache, LRUCache, TTLCache, func
@@ -466,9 +465,9 @@ def _utxo_from_ogmios_result(self, result) -> UTxO:
         script = output.get("script", None)
         if script:
             if "plutus:v2" in script:
-                script = PlutusV2Script(cbor2.loads(bytes.fromhex(script["plutus:v2"])))
+                script = PlutusV2Script(bytes.fromhex(script["plutus:v2"]))
             elif "plutus:v1" in script:
-                script = PlutusV1Script(cbor2.loads(bytes.fromhex(script["plutus:v1"])))
+                script = PlutusV1Script(bytes.fromhex(script["plutus:v1"]))
             else:
                 raise ValueError("Unknown plutus script type")
         datum_hash = (

pycardano/cip/cip8.py

@@ -1,5 +1,6 @@
 from typing import Optional, Union
 
+from cbor2 import CBORTag, dumps
 from cose.algorithms import EdDSA
 from cose.headers import KID, Algorithm
 from cose.keys import CoseKey
@@ -10,13 +11,15 @@
 from cose.messages import CoseMessage, Sign1Message
 
 from pycardano.address import Address
+from pycardano.crypto import BIP32ED25519PublicKey
 from pycardano.key import (
+    ExtendedSigningKey,
+    ExtendedVerificationKey,
     PaymentVerificationKey,
     SigningKey,
     StakeExtendedSigningKey,
     StakeSigningKey,
     StakeVerificationKey,
-    VerificationKey,
 )
 from pycardano.network import Network
 
@@ -25,7 +28,7 @@
 
 def sign(
     message: str,
-    signing_key: SigningKey,
+    signing_key: Union[ExtendedSigningKey, SigningKey],
     attach_cose_key: bool = False,
     network: Network = Network.MAINNET,
 ) -> Union[str, dict]:
@@ -45,7 +48,9 @@ def sign(
     """
 
     # derive the verification key
-    verification_key = VerificationKey.from_signing_key(signing_key)
+    verification_key = signing_key.to_verification_key()
+    if isinstance(verification_key, ExtendedVerificationKey):
+        verification_key = verification_key.to_non_extended()
 
     if isinstance(signing_key, StakeSigningKey) or isinstance(
         signing_key, StakeExtendedSigningKey
@@ -85,7 +90,20 @@ def sign(
 
     msg.key = cose_key  # attach the key to the message
 
-    encoded = msg.encode()
+    if isinstance(signing_key, ExtendedSigningKey):
+        _message = [
+            msg.phdr_encoded,
+            msg.uhdr_encoded,
+            msg.payload,
+            signing_key.sign(msg._sig_structure),
+        ]
+
+        encoded = dumps(
+            CBORTag(msg.cbor_tag, _message), default=msg._custom_cbor_encoder
+        )
+
+    else:
+        encoded = msg.encode()
 
     # turn the enocded message into a hex string and remove the first byte
     # which is always "d2"
@@ -108,7 +126,8 @@ def sign(
 
 
 def verify(
-    signed_message: Union[str, dict], attach_cose_key: Optional[bool] = None
+    signed_message: Union[str, dict],
+    attach_cose_key: Optional[bool] = None,
 ) -> dict:
     """Verify the signature of a COSESign1 message and decode its contents following CIP-0008.
     Supports messages signed by browser wallets or `Message.sign()`.
@@ -175,7 +194,16 @@ def verify(
     # attach the key to the decoded message
     decoded_message.key = cose_key
 
-    signature_verified = decoded_message.verify_signature()
+    if len(verification_key) > 32:
+        vk = BIP32ED25519PublicKey(
+            public_key=verification_key[:32], chain_code=verification_key[32:]
+        )
+        vk.verify(
+            signature=decoded_message.signature, message=decoded_message._sig_structure
+        )
+        signature_verified = True
+    else:
+        signature_verified = decoded_message.verify_signature()
 
     message = decoded_message.payload.decode("utf-8")
 

pycardano/metadata.py

@@ -41,6 +41,7 @@ def _validate_type_and_size(data):
                 if len(data) > self.MAX_ITEM_SIZE:
                     raise InvalidArgumentException(
                         f"The size of {data} exceeds {self.MAX_ITEM_SIZE} bytes."
+                        "Use pycardano.serialization.ByteString for long bytes."
                     )
             elif isinstance(data, str):
                 if len(data.encode("utf-8")) > self.MAX_ITEM_SIZE:

pycardano/plutus.py

@@ -6,18 +6,20 @@
 import json
 from dataclasses import dataclass, field, fields
 from enum import Enum
-from typing import Any, ClassVar, Optional, Type, Union
+from hashlib import sha256
+from typing import Any, Optional, Type, Union
 
 import cbor2
 from cbor2 import CBORTag
 from nacl.encoding import RawEncoder
 from nacl.hash import blake2b
 
-from pycardano.exception import DeserializeException
+from pycardano.exception import DeserializeException, InvalidArgumentException
 from pycardano.hash import DATUM_HASH_SIZE, SCRIPT_HASH_SIZE, DatumHash, ScriptHash
 from pycardano.nativescript import NativeScript
 from pycardano.serialization import (
     ArrayCBORSerializable,
+    ByteString,
     CBORSerializable,
     DictCBORSerializable,
     IndefiniteList,
@@ -44,9 +46,16 @@
     "datum_hash",
     "plutus_script_hash",
     "script_hash",
+    "Unit",
 ]
 
 
+# taken from https://stackoverflow.com/a/13624858
+class classproperty(property):
+    def __get__(self, owner_self, owner_cls):
+        return self.fget(owner_cls)
+
+
 class CostModels(DictCBORSerializable):
     KEY_TYPE = int
     VALUE_TYPE = dict
@@ -460,18 +469,43 @@ class will reduce the complexity of serialization and deserialization tremendous
         >>> assert test == Test.from_cbor("d87a9f187b43333231ff")
     """
 
-    CONSTR_ID: ClassVar[int] = 0
-    """Constructor ID of this plutus data.
-       It is primarily used by Plutus core to reconstruct a data structure from serialized CBOR bytes."""
+    MAX_BYTES_SIZE = 64
+
+    @classproperty
+    def CONSTR_ID(cls):
+        """
+        Constructor ID of this plutus data.
+        It is primarily used by Plutus core to reconstruct a data structure from serialized CBOR bytes.
+        The default implementation is an almost unique, deterministic constructor ID in the range 1 - 2^32 based
+        on class attributes, types and class name.
+        """
+        k = f"_CONSTR_ID_{cls.__name__}"
+        if not hasattr(cls, k):
+            det_string = (
+                cls.__name__
+                + "*"
+                + "*".join([f"{f.name}~{f.type}" for f in fields(cls)])
+            )
+            det_hash = sha256(det_string.encode("utf8")).hexdigest()
+            setattr(cls, k, int(det_hash, 16) % 2**32)
+
+        return getattr(cls, k)
 
     def __post_init__(self):
-        valid_types = (PlutusData, dict, IndefiniteList, int, bytes)
+        valid_types = (PlutusData, dict, IndefiniteList, int, ByteString, bytes)
         for f in fields(self):
             if inspect.isclass(f.type) and not issubclass(f.type, valid_types):
                 raise TypeError(
                     f"Invalid field type: {f.type}. A field in PlutusData should be one of {valid_types}"
                 )
 
+            data = getattr(self, f.name)
+            if isinstance(data, bytes) and len(data) > 64:
+                raise InvalidArgumentException(
+                    f"The size of {data} exceeds {self.MAX_BYTES_SIZE} bytes. "
+                    "Use pycardano.serialization.ByteString for long bytes."
+                )
+
     def to_shallow_primitive(self) -> CBORTag:
         primitives: Primitive = super().to_shallow_primitive()
         if primitives:
@@ -529,6 +563,8 @@ def _dfs(obj):
                 return {"int": obj}
             elif isinstance(obj, bytes):
                 return {"bytes": obj.hex()}
+            elif isinstance(obj, ByteString):
+                return {"bytes": obj.value.hex()}
             elif isinstance(obj, IndefiniteList) or isinstance(obj, list):
                 return {"list": [_dfs(item) for item in obj]}
             elif isinstance(obj, dict):
@@ -643,7 +679,10 @@ def _dfs(obj):
                 elif "int" in obj:
                     return obj["int"]
                 elif "bytes" in obj:
-                    return bytes.fromhex(obj["bytes"])
+                    if len(obj["bytes"]) > 64:
+                        return ByteString(bytes.fromhex(obj["bytes"]))
+                    else:
+                        return bytes.fromhex(obj["bytes"])
                 elif "list" in obj:
                     return IndefiniteList([_dfs(item) for item in obj["list"]])
                 else:
@@ -820,3 +859,10 @@ def script_hash(script: ScriptType) -> ScriptHash:
         )
     else:
         raise TypeError(f"Unexpected script type: {type(script)}")
+
+
+@dataclass
+class Unit(PlutusData):
+    """The default "Unit type" with a 0 constructor ID"""
+
+    CONSTR_ID = 0

pycardano/serialization.py

@@ -60,6 +60,22 @@ class IndefiniteFrozenList(FrozenList, IndefiniteList):  # type: ignore
     pass
 
 
+@dataclass
+class ByteString:
+    value: bytes
+
+    def __hash__(self):
+        return hash(self.value)
+
+    def __eq__(self, other: object):
+        if isinstance(other, ByteString):
+            return self.value == other.value
+        elif isinstance(other, bytes):
+            return self.value == other
+        else:
+            return False
+
+
 @dataclass
 class RawCBOR:
     """A wrapper class for bytes that represents a CBOR value."""
@@ -160,6 +176,7 @@ def default_encoder(
     assert isinstance(
         value,
         (
+            ByteString,
             CBORSerializable,
             IndefiniteList,
             RawCBOR,
@@ -178,6 +195,15 @@ def default_encoder(
         for item in value:
             encoder.encode(item)
         encoder.write(b"\xff")
+    elif isinstance(value, ByteString):
+        if len(value.value) > 64:
+            encoder.write(b"\x5f")
+            for i in range(0, len(value.value), 64):
+                imax = min(i + 64, len(value.value))
+                encoder.encode(value.value[i:imax])
+            encoder.write(b"\xff")
+        else:
+            encoder.encode(value.value)
     elif isinstance(value, RawCBOR):
         encoder.write(value.cbor)
     elif isinstance(value, FrozenList):

pycardano/txbuilder.py

@@ -1148,6 +1148,13 @@ def _add_collateral_input(cur_total, candidate_inputs):
             )
             _add_collateral_input(tmp_val, sorted_inputs)
 
+            if tmp_val.coin < collateral_amount:
+                sorted_inputs = sorted(
+                    self.potential_inputs,
+                    key=lambda i: (len(i.output.to_cbor_hex()), -i.output.amount.coin),
+                )
+                _add_collateral_input(tmp_val, sorted_inputs)
+
             if tmp_val.coin < collateral_amount:
                 sorted_inputs = sorted(
                     self.context.utxos(collateral_return_address),