diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index a29064fa..7b10c7fa 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -7,6 +7,11 @@ on: branches: ["main", "chang"] types: [opened, reopened, edited, synchronize] +env: + # Keep in sync with the generator recorded in the first line of poetry.lock. + # An older Poetry rejects the lock file's content hash; a newer one may rewrite it. + POETRY_VERSION: "2.4.2" + jobs: build: @@ -19,7 +24,7 @@ jobs: steps: - uses: actions/checkout@v4 - name: Install poetry - run: pipx install poetry + run: pipx install poetry==${{ env.POETRY_VERSION }} - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v5 with: @@ -52,7 +57,7 @@ jobs: steps: - uses: actions/checkout@v4 - name: Install poetry - run: pipx install poetry + run: pipx install poetry==${{ env.POETRY_VERSION }} - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v5 with: diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index c451ff40..927066d1 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -4,6 +4,10 @@ on: tags: - "v[0-9]+.[0-9]+.[0-9]+" # Push events to matching *, i.e. 1.0, 20.15.10 +env: + # Keep in sync with the generator recorded in the first line of poetry.lock. + POETRY_VERSION: "2.4.2" + jobs: publish: strategy: @@ -15,7 +19,7 @@ jobs: steps: - uses: actions/checkout@v3 - name: Install poetry - run: pipx install poetry + run: pipx install poetry==${{ env.POETRY_VERSION }} - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v5 with: diff --git a/pycardano/cip/cip8.py b/pycardano/cip/cip8.py index 1595221d..5d83f494 100644 --- a/pycardano/cip/cip8.py +++ b/pycardano/cip/cip8.py @@ -12,6 +12,7 @@ from pycardano.address import Address from pycardano.crypto import BIP32ED25519PublicKey +from pycardano.exception import DeserializeException from pycardano.key import ( ExtendedSigningKey, ExtendedVerificationKey, @@ -162,7 +163,15 @@ def verify( signed_message, str ), "signed_message must be a hex string at this point" decoded = loads(bytes.fromhex("d2" + signed_message)) - assert isinstance(decoded, CBORTag) and decoded.tag == Sign1Message.cbor_tag + if ( + not isinstance(decoded, CBORTag) + or decoded.tag != Sign1Message.cbor_tag + or not isinstance(decoded.value, list) + ): + raise DeserializeException( + f"Expected a COSE_Sign1 message tagged {Sign1Message.cbor_tag} with an " + f"array body, got {decoded!r}." + ) decoded_message = Sign1Message.from_cose_obj( decoded.value, allow_unknown_attributes=True ) diff --git a/pycardano/serialization.py b/pycardano/serialization.py index 2422b3f3..82ae0c6c 100644 --- a/pycardano/serialization.py +++ b/pycardano/serialization.py @@ -47,8 +47,14 @@ __all__ = [ "default_encoder", + "Indefinite", "IndefiniteList", "IndefiniteFrozenList", + "IndefiniteString", + "IndefiniteByteString", + "IndefiniteTextString", + "IndefiniteDict", + "IndefiniteFrozenDict", "Primitive", "CBORBase", "CBORSerializable", @@ -70,7 +76,16 @@ def _identity(x): return x -class IndefiniteList(UserList): +class Indefinite: + """Marker for a value whose CBOR encoding used an indefinite-length header. + + The choice between a definite and an indefinite header is part of the encoding, + and therefore part of every hash computed over it, so PyCardano carries it on the + decoded value instead of discarding it. + """ + + +class IndefiniteList(Indefinite, UserList): def __init__(self, li: Primitive): # type: ignore if isinstance(li, FrozenList): super().__init__(list(li)) # type: ignore @@ -82,6 +97,159 @@ class IndefiniteFrozenList(FrozenList, IndefiniteList): # type: ignore pass +class IndefiniteString(Indefinite): + """Shared chunk bookkeeping for the two indefinite-length string types. + + An indefinite string is written as a header, a run of definite-length chunks and + a break marker. The chunk lengths are part of the encoding, so they are recorded + on decode and replayed on encode rather than renormalised. + """ + + CBOR_HEADER: ClassVar[bytes] + chunks: tuple + + @classmethod + def _default_chunks(cls, payload: bytes) -> tuple: + """Chunking to use when a value is constructed without explicit lengths.""" + raise NotImplementedError + + @classmethod + def _resolve_chunks(cls, payload: bytes, chunks: Optional[Iterable[int]]) -> tuple: + resolved = cls._default_chunks(payload) if chunks is None else tuple(chunks) + if sum(resolved) != len(payload): + raise ValueError( + f"Chunk lengths {resolved} do not add up to the {len(payload)} byte " + f"value they describe." + ) + return resolved + + def cbor_payload(self) -> bytes: + """The bytes the chunk lengths are measured against.""" + raise NotImplementedError + + @staticmethod + def _as_chunk(raw: bytes) -> Any: + """One slice of the payload, in the form the encoder should write.""" + raise NotImplementedError + + def cbor_chunks(self) -> Iterable[Any]: + payload = self.cbor_payload() + start = 0 + for length in self.chunks: + yield self._as_chunk(payload[start : start + length]) + start += length + + +class IndefiniteByteString(IndefiniteString, bytes): + """A byte string that was CBOR encoded as indefinite length chunks.""" + + CBOR_HEADER = b"\x5f" + + def __new__(cls, value: bytes, chunks: Optional[Iterable[int]] = None): + self = super().__new__(cls, value) + self.chunks = cls._resolve_chunks(bytes(value), chunks) + return self + + @classmethod + def _default_chunks(cls, payload: bytes) -> tuple: + # Cardano splits byte strings longer than 64 bytes into 64 byte chunks. + return tuple( + min(64, len(payload) - start) for start in range(0, len(payload), 64) + ) + + def cbor_payload(self) -> bytes: + return bytes(self) + + @staticmethod + def _as_chunk(raw: bytes) -> bytes: + return raw + + def __repr__(self): + return f"IndefiniteByteString({bytes(self)!r}, chunks={self.chunks})" + + def __deepcopy__(self, memo): + return IndefiniteByteString(bytes(self), self.chunks) + + def __reduce__(self): + return (IndefiniteByteString, (bytes(self), self.chunks)) + + +class IndefiniteTextString(IndefiniteString, str): + """A text string that was CBOR encoded as indefinite length chunks. + + Chunk lengths count UTF-8 bytes, matching how the chunks appear on the wire. + """ + + CBOR_HEADER = b"\x7f" + + def __new__(cls, value: str, chunks: Optional[Iterable[int]] = None): + self = super().__new__(cls, value) + self.chunks = cls._resolve_chunks(str(value).encode("utf-8"), chunks) + return self + + @classmethod + def _default_chunks(cls, payload: bytes) -> tuple: + return (len(payload),) if payload else () + + def cbor_payload(self) -> bytes: + return str(self).encode("utf-8") + + @staticmethod + def _as_chunk(raw: bytes) -> str: + return raw.decode("utf-8") + + def __repr__(self): + return f"IndefiniteTextString({str(self)!r}, chunks={self.chunks})" + + def __deepcopy__(self, memo): + return IndefiniteTextString(str(self), self.chunks) + + def __reduce__(self): + return (IndefiniteTextString, (str(self), self.chunks)) + + +class IndefiniteDict(Indefinite, dict): + """A map that was CBOR encoded with an indefinite length header.""" + + def __repr__(self): + return f"IndefiniteDict({dict(self)!r})" + + def __deepcopy__(self, memo): + return type(self)( + {deepcopy(k, memo): deepcopy(v, memo) for k, v in self.items()} + ) + + +class IndefiniteFrozenDict(IndefiniteDict): + """Hashable counterpart of :class:`IndefiniteDict`, used for map keys.""" + + def __hash__(self): # type: ignore[override] + return hash(frozenset(self.items())) + + def _immutable(self, *args, **kwargs): + raise TypeError(f"{type(self).__name__} is immutable") + + __setitem__ = _immutable + __delitem__ = _immutable + clear = _immutable + pop = _immutable # type: ignore[assignment] + popitem = _immutable # type: ignore[assignment] + setdefault = _immutable # type: ignore[assignment] + update = _immutable # type: ignore[assignment] + + +# cbor2 looks encoders up by exact type, so every concrete class has to be claimed +# for its representation to survive; the base classes above are not enough. +_INDEFINITE_TYPES = ( + IndefiniteList, + IndefiniteFrozenList, + IndefiniteByteString, + IndefiniteTextString, + IndefiniteDict, + IndefiniteFrozenDict, +) + + @dataclass class ByteString: value: bytes @@ -194,6 +362,14 @@ def wrapper(cls, value: Primitive): CBORBase = TypeVar("CBORBase", bound="CBORSerializable") +def _write_indefinite_string(encoder: CBOREncoder, value: IndefiniteString) -> None: + """Write an indefinite-length string as the exact chunks it was encoded with.""" + encoder.write(value.CBOR_HEADER) + for chunk in value.cbor_chunks(): + encoder.encode(chunk) + encoder.write(b"\xff") + + def default_encoder( encoder: CBOREncoder, value: Union[CBORSerializable, IndefiniteList] ): @@ -203,10 +379,9 @@ def default_encoder( ( ByteString, CBORSerializable, - IndefiniteList, + Indefinite, RawCBOR, FrozenList, - IndefiniteFrozenList, FrozenDict, ), ), ( @@ -219,15 +394,25 @@ def default_encoder( for item in value: encoder.encode(item) encoder.write(b"\xff") + elif isinstance(value, IndefiniteString): + _write_indefinite_string(encoder, value) + elif isinstance(value, IndefiniteDict): + encoder.write(b"\xbf") + for item_key, item_value in value.items(): + encoder.encode(item_key) + encoder.encode(item_value) + 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") + inner = value.value + if isinstance(inner, IndefiniteByteString): + # The wrapped value remembers how it arrived, so reuse those chunks. + _write_indefinite_string(encoder, inner) + elif len(inner) > 64: + # Too long for a definite byte string, so fall back to the 64 byte + # chunks Cardano uses. + _write_indefinite_string(encoder, IndefiniteByteString(inner)) else: - encoder.encode(value.value) + encoder.encode(inner) elif isinstance(value, RawCBOR): encoder.write(value.cbor) elif isinstance(value, FrozenList): @@ -242,8 +427,10 @@ def dumps(obj: Any, **kwargs: Any) -> bytes: """Encode CBOR while preserving PyCardano's indefinite-list types.""" encoders = dict(kwargs.pop("encoders", {})) encoder = kwargs.get("default", default_encoder) - encoders.setdefault(IndefiniteList, encoder) - encoders.setdefault(IndefiniteFrozenList, encoder) + for indefinite_type in _INDEFINITE_TYPES: + # cbor2 encodes these natively as their definite-length base types, so each + # one has to be claimed explicitly for the representation to survive. + encoders.setdefault(indefinite_type, encoder) return cbor2.dumps(obj, encoders=encoders, **kwargs) @@ -256,6 +443,17 @@ class _CBORShape: major_type: int indefinite: bool = False children: tuple = () + # Byte length of each chunk, for indefinite-length strings only. + chunks: tuple = () + + +# Major types that decode to a scalar. They carry no representation choice worth +# restoring, so every occurrence can share one shape instance instead of allocating +# a new one per decoded item. +_SCALAR_MAJOR_TYPES = frozenset((0, 1, 2, 3, 7)) +_SCALAR_SHAPES = { + major_type: _CBORShape(major_type) for major_type in _SCALAR_MAJOR_TYPES +} def _read_argument(payload: memoryview, offset: int, subtype: int) -> tuple[int, int]: @@ -277,15 +475,25 @@ def _scan_cbor(payload: memoryview, offset: int = 0) -> tuple[_CBORShape, int]: argument, offset = _read_argument(payload, offset, subtype) if major_type in (0, 1, 7): - return _CBORShape(major_type), offset + return _SCALAR_SHAPES[major_type], offset if major_type in (2, 3): if argument is None: + # Chunks of an indefinite string are always definite strings of the same + # major type, so their headers can be read directly. + chunk_lengths: List[int] = [] while payload[offset] != 0xFF: - _, offset = _scan_cbor(payload, offset) - return _CBORShape(major_type, indefinite=True), offset + 1 + chunk_subtype = payload[offset] % 32 + offset += 1 + chunk_length, offset = _read_argument(payload, offset, chunk_subtype) + chunk_lengths.append(chunk_length) + offset += chunk_length + return ( + _CBORShape(major_type, True, (), tuple(chunk_lengths)), + offset + 1, + ) - return _CBORShape(major_type), offset + argument + return _SCALAR_SHAPES[major_type], offset + argument if major_type == 4: array_children: List[_CBORShape] = [] @@ -322,7 +530,22 @@ def _scan_cbor(payload: memoryview, offset: int = 0) -> tuple[_CBORShape, int]: def _apply_cbor_shape(value: Any, shape: _CBORShape, immutable: bool = False) -> Any: - if shape.major_type == 4: + major_type = shape.major_type + + if major_type in _SCALAR_MAJOR_TYPES: + if shape.indefinite: + if major_type == 2 and isinstance(value, (bytes, bytearray)): + return IndefiniteByteString(bytes(value), shape.chunks) + if major_type == 3 and isinstance(value, str): + return IndefiniteTextString(value, shape.chunks) + return value + + if major_type == 4: + # A semantic decoder may have replaced the array with an unrelated object, so + # only walk into it when the decoded value still looks like the array we scanned. + if not isinstance(value, (list, tuple)) or len(value) != len(shape.children): + return value + array_items = [ _apply_cbor_shape(item, item_shape, immutable) for item, item_shape in zip(value, shape.children) @@ -336,7 +559,20 @@ def _apply_cbor_shape(value: Any, shape: _CBORShape, immutable: bool = False) -> return tuple(array_items) if immutable else array_items - if shape.major_type == 5: + if major_type == 5: + if not isinstance(value, (dict, FrozenDict)): + return value + + if len(value) != len(shape.children): + # Entries were lost on the way in, which is what duplicate keys do: a + # decoded map holds one entry per distinct key, so the scanned shapes can + # no longer be matched to the values they came from. The keys are still + # correct, so hand the map back as decoded rather than guessing, which + # would attribute a shape to the wrong value. The representation of what + # is nested inside is not recoverable either way, which is why loads() + # rejects duplicate keys unless a caller opts in. + return value + paired_items = zip(value.items(), shape.children) mapped_items = { _apply_cbor_shape(key, key_shape, immutable=True): _apply_cbor_shape( @@ -344,22 +580,50 @@ def _apply_cbor_shape(value: Any, shape: _CBORShape, immutable: bool = False) -> ) for (key, item), (key_shape, item_shape) in paired_items } + if shape.indefinite: + return ( + IndefiniteFrozenDict(mapped_items) + if immutable + else IndefiniteDict(mapped_items) + ) + return FrozenDict(mapped_items) if immutable else mapped_items - if shape.major_type == 6 and isinstance(value, CBORTag): - child = _apply_cbor_shape(value.value, shape.children[0], immutable) - return CBORTag(value.tag, child) + if major_type == 6: + if isinstance(value, CBORTag): + child = _apply_cbor_shape(value.value, shape.children[0], immutable) + return CBORTag(value.tag, child) + + # A semantic decoder consumed the tag and returned its content directly (the + # self-describe tag is decoded this way). The scanned child shape describes + # what is left, and the guards above ignore it if the decoder returned + # something else entirely. + return _apply_cbor_shape(value, shape.children[0], immutable) return value def loads(payload: bytes, **kwargs: Any) -> Any: - """Decode CBOR while preserving order and indefinite-length arrays.""" + """Decode CBOR while preserving order and indefinite-length containers. + + Duplicate map keys are rejected by default. They collapse into a single entry on + decode, which loses data and makes the original encoding impossible to reproduce. + Pass ``allow_duplicate_keys=True`` to decode such a payload anyway; the keys come + back correct, but the representation of that map and anything nested inside it is + not preserved, so re-encoding will not reproduce the input bytes. + """ semantic_decoders = dict(kwargs.pop("semantic_decoders", {})) semantic_decoders.setdefault(258, _decode_ordered_set) + # Duplicate keys collapse into a single entry on decode, which loses data and + # makes the original encoding impossible to reproduce. Reject them by default. + kwargs.setdefault("allow_duplicate_keys", False) value = cbor2.loads(payload, semantic_decoders=semantic_decoders, **kwargs) shape, offset = _scan_cbor(memoryview(payload)) - assert offset == len(payload) + if offset != len(payload): + raise DeserializeException( + f"Trailing CBOR data: {len(payload) - offset} unexpected byte(s) follow " + f"the top level value." + ) return _apply_cbor_shape(value, shape) diff --git a/test/pycardano/test_cip8.py b/test/pycardano/test_cip8.py index 778af0b2..c656c5c2 100644 --- a/test/pycardano/test_cip8.py +++ b/test/pycardano/test_cip8.py @@ -1,5 +1,8 @@ +import pytest + from pycardano.cip.cip8 import sign, verify from pycardano.crypto.bip32 import BIP32ED25519PrivateKey, HDWallet +from pycardano.exception import DeserializeException from pycardano.key import ( ExtendedSigningKey, ExtendedVerificationKey, @@ -48,6 +51,12 @@ }""") +def test_verify_non_cose_message_raises(): + """A payload that is not a COSE_Sign1 array must fail with a clear error.""" + with pytest.raises(DeserializeException, match="COSE_Sign1"): + verify("01") + + def test_verify_message(): signed_message = "845869a3012704582060545b786d3a6f903158e35aae9b86548a99bc47d4b0a6f503ab5e78c1a9bbfc6761646472657373583900ddba3ad76313825f4f646f5aa6d323706653bda40ec1ae55582986a463e661768b92deba45b5ada4ab9e7ffd17ed3051b2e03500e0542e9aa166686173686564f452507963617264616e6f20697320636f6f6c2e58403b09cbae8d272ff94befd28cc04b152aea3c1633caffb4924a8a8c45be3ba6332a76d9f2aba833df53803286d32a5ee700990b79a0e86fab3cccdbfd37ce250f" diff --git a/test/pycardano/test_plutus.py b/test/pycardano/test_plutus.py index 49eb1038..51878481 100644 --- a/test/pycardano/test_plutus.py +++ b/test/pycardano/test_plutus.py @@ -497,6 +497,24 @@ class A(PlutusData): ), "Long metadata bytestring is encoded incorrectly." +def test_raw_plutus_data_preserves_long_bytes_chunking(): + """Regression for https://github.com/Python-Cardano/pycardano/issues/507 + + The ledger rejects a definite-length bytestring longer than 64 bytes, so a datum + holding one is encoded in chunks. TransactionBuilder deep-copies its outputs and + RawPlutusData.__deepcopy__ goes through CBOR, so the chunking has to survive both + a copy and a round trip, or the builder emits a datum the node refuses. + """ + datum = RawPlutusData.from_dict( + {"constructor": 0, "fields": [{"bytes": "ab" * 182}]} + ) + encoded = datum.to_cbor() + + assert bytes.fromhex("5f5840") in encoded, "long bytes were not chunked" + assert copy.deepcopy(datum).to_cbor() == encoded + assert RawPlutusData.from_cbor(encoded).to_cbor() == encoded + + def test_plutus_raw_plutus_data(): @dataclass diff --git a/test/pycardano/test_serialization.py b/test/pycardano/test_serialization.py index 4daa4be7..bd5f7350 100644 --- a/test/pycardano/test_serialization.py +++ b/test/pycardano/test_serialization.py @@ -4,6 +4,8 @@ from collections import defaultdict, deque from copy import deepcopy from dataclasses import dataclass, field +from decimal import Decimal +from fractions import Fraction from test.pycardano.util import check_two_way_cbor from typing import ( Any, @@ -48,7 +50,10 @@ CBORSerializable, CodedSerializable, DictCBORSerializable, + Indefinite, + IndefiniteByteString, IndefiniteList, + IndefiniteTextString, MapCBORSerializable, NonEmptyOrderedSet, OrderedSet, @@ -1171,6 +1176,152 @@ def test_indefinite_array_item_count_byte_exact_round_trip(length): assert dumps(decoded) == encoded +def test_duplicate_map_keys_are_rejected(): + """A collapsed duplicate key cannot be re-encoded to the original bytes.""" + # {1: [2], 1: [2 indefinite]} -- decoding keeps one entry, losing the other. + encoded = bytes.fromhex("a2018102019f02ff") + + with pytest.raises(cbor2.CBORDecodeError): + loads(encoded) + + +def test_duplicate_map_keys_allowed_explicitly_keep_their_keys(): + """Opting into duplicate keys yields the decoded map, not a mis-shaped one. + + The dropped entry makes the original encoding unrecoverable, so the surviving + value keeps no array shape and re-encoding does not reproduce the input. + """ + # {1: [2], 1: [2 indefinite]} -- one entry survives, carrying the later value. + encoded = bytes.fromhex("a2018102019f02ff") + + decoded = loads(encoded, allow_duplicate_keys=True) + + assert decoded == {1: [2]} + assert dumps(decoded) == bytes.fromhex("a1018102") + + +def test_trailing_bytes_raise_deserialize_exception(): + with pytest.raises(DeserializeException, match="Trailing CBOR data"): + loads(bytes.fromhex("0102")) + + +def test_self_describe_tag_content_is_mutable(): + """cbor2 6 decodes tag content as immutable; the shape pass restores it.""" + decoded = loads(bytes.fromhex("d9d9f7820102")) + + assert isinstance(decoded, list) + assert decoded == [1, 2] + + +def test_self_describe_tag_preserves_indefinite_array(): + decoded = loads(bytes.fromhex("d9d9f79f0102ff")) + + assert isinstance(decoded, IndefiniteList) + assert dumps(decoded) == bytes.fromhex("9f0102ff") + + +@pytest.mark.parametrize( + ("cbor_hex", "expected"), + [ + pytest.param("d81e820102", Fraction(1, 2), id="tag30-fraction"), + pytest.param("c249010000000000000000", 2**64, id="tag2-bignum"), + pytest.param("c48221196ab3", Decimal("273.15"), id="tag4-decimal"), + ], +) +def test_consumed_tags_keep_their_decoded_value(cbor_hex, expected): + """A semantic decoder replaces the tag content, so no array shape applies.""" + assert loads(bytes.fromhex(cbor_hex)) == expected + + +@pytest.mark.parametrize( + "cbor_hex", + [ + pytest.param("5f4101ff", id="indefinite-byte-string"), + pytest.param("5fff", id="empty-indefinite-byte-string"), + pytest.param("5f41014102ff", id="multi-chunk-byte-string"), + pytest.param("7f6161ff", id="indefinite-text-string"), + pytest.param("7f62c3a96162ff", id="indefinite-text-string-multibyte"), + pytest.param("bf0102ff", id="indefinite-map"), + pytest.param("bfff", id="empty-indefinite-map"), + pytest.param("bf01bf0203ffff", id="nested-indefinite-maps"), + pytest.param("a15f4101ff02", id="indefinite-byte-string-as-map-key"), + pytest.param("a1bf0102ff03", id="indefinite-map-as-map-key"), + pytest.param("81bf0102ff", id="indefinite-map-in-array"), + pytest.param("d8799f5f4101ffff", id="chunked-byte-string-in-plutus-tag"), + ], +) +def test_indefinite_string_and_map_byte_exact_round_trip(cbor_hex): + """Indefinite strings and maps must survive a decode/encode cycle unchanged.""" + encoded = bytes.fromhex(cbor_hex) + + assert dumps(loads(encoded)) == encoded + + +def test_chunked_datum_keeps_its_ledger_hash(): + """A >64 byte Plutus byte string arrives chunked; re-encoding must not move it.""" + chunked = bytes.fromhex( + "d8799f5f5840" + bytes(range(64)).hex() + "48" + "00" * 8 + "ffff" + ) + + restored = RawPlutusData.from_cbor(chunked) + + assert restored.to_cbor() == chunked + + +def test_indefinite_byte_string_behaves_like_bytes(): + decoded = loads(bytes.fromhex("5f41014102ff")) + + assert isinstance(decoded, bytes) + assert decoded == b"\x01\x02" + assert hash(decoded) == hash(b"\x01\x02") + assert decoded.chunks == (1, 1) + + +@pytest.mark.parametrize( + "cbor_hex", + [ + pytest.param("9f01ff", id="list"), + pytest.param("5f4101ff", id="byte-string"), + pytest.param("7f6161ff", id="text-string"), + pytest.param("bf0102ff", id="map"), + ], +) +def test_every_indefinite_type_shares_the_marker_base(cbor_hex): + """default_encoder dispatches on Indefinite, so each type must carry it.""" + assert isinstance(loads(bytes.fromhex(cbor_hex)), Indefinite) + + +def test_indefinite_map_key_is_hashable_and_immutable(): + decoded = loads(bytes.fromhex("a1bf0102ff03")) + key = next(iter(decoded)) + + assert isinstance(key, dict) + assert key == {1: 2} + with pytest.raises(TypeError): + key[3] = 4 + + +def test_indefinite_strings_survive_deep_copy(): + byte_string = IndefiniteByteString(b"\x01\x02", (1, 1)) + text_string = IndefiniteTextString("ab", (1, 1)) + + assert deepcopy(byte_string).chunks == (1, 1) + assert deepcopy(text_string).chunks == (1, 1) + + +def test_chunk_lengths_must_match_the_value(): + with pytest.raises(ValueError, match="do not add up"): + IndefiniteByteString(b"\x01\x02", (1,)) + + +def test_byte_string_field_reuses_the_chunks_it_was_decoded_with(): + """ByteString defaults to 64 byte chunks, but must honour an unusual encoding.""" + original = bytes.fromhex("5f41014102ff") + wrapped = ByteString(loads(original)) + + assert dumps(wrapped, default=default_encoder) == original + + def test_ordered_set_as_key_in_dict_indefinite_list(): a = NonEmptyOrderedSet(IndefiniteList([1, 2, 3])) diff --git a/test/pycardano/test_transaction.py b/test/pycardano/test_transaction.py index 3049c000..698a5f7d 100644 --- a/test/pycardano/test_transaction.py +++ b/test/pycardano/test_transaction.py @@ -15,7 +15,13 @@ from pycardano.hash import SCRIPT_HASH_SIZE, ScriptHash, TransactionId from pycardano.key import PaymentKeyPair, PaymentSigningKey, VerificationKey from pycardano.nativescript import ScriptPubkey -from pycardano.plutus import PlutusData, PlutusV1Script, PlutusV2Script, datum_hash +from pycardano.plutus import ( + PlutusData, + PlutusV1Script, + PlutusV2Script, + RawPlutusData, + datum_hash, +) from pycardano.transaction import ( Asset, AssetName, @@ -745,6 +751,53 @@ def test_sign_dex_transaction_with_raw_plutus_data_inline_datum(): nacl_vk.verify(body_hash, signature) +def test_sign_round_tripped_transaction_with_chunked_datum_bytestring(): + # Regression for https://github.com/python-cardano/pycardano/issues/402 + # + # The fixture above covers a datum whose byte strings all fit in a single + # definite-length chunk. Real DEX and escrow datums carry longer values - a COSE + # signature, a long hash - and the ledger rejects a definite-length byte string + # over 64 bytes, so those arrive split into chunks. Decoding used to drop the chunk + # boundaries, so re-encoding the body moved its hash and any signature computed + # over the original body no longer verified. + address = Address.from_primitive( + "addr_test1qqqt0pru382hy9vjlsxv3ye02z50sfvt8xunscg5pgden77z73dpdfng2ctw2ekqp" + "lqgrljelz7h4dneac27nn3qx3rqqpavzj" + ) + datum = RawPlutusData.from_dict( + {"constructor": 0, "fields": [{"bytes": "ab" * 182}]} + ) + body = TransactionBody( + inputs=[TransactionInput(TransactionId(b"\x11" * 32), 0)], + outputs=[ + TransactionOutput(address, Value(5000000), datum=datum, post_alonzo=True) + ], + fee=200000, + ) + + # A wallet exports the unsigned transaction; the signature is made over the body + # hash as it stands at that point. + unsigned_cbor_hex = Transaction(body, TransactionWitnessSet()).to_cbor_hex() + sk = PaymentSigningKey.from_json("""{ + "type": "GenesisUTxOSigningKey_ed25519", + "description": "Genesis Initial UTxO Signing Key", + "cborHex": "5820093be5cd3987d0c9fd8854ef908f7746b69e2d73320db6dc0f780d81585b84c2" + }""") + signature = sk.sign(body.hash()) + + restored = Transaction.from_cbor(unsigned_cbor_hex) + + assert bytes.fromhex("5f5840") in bytes.fromhex( + unsigned_cbor_hex + ), "long bytes were not chunked" + assert restored.to_cbor_hex() == unsigned_cbor_hex + assert restored.transaction_body.hash() == body.hash() + + # The signature made before the round trip must still verify afterwards. + nacl_vk = NACLSigningKey(sk.payload).verify_key + nacl_vk.verify(restored.transaction_body.hash(), signature) + + def test_decode_byron_transaction(): tx_cbor_hex = """83a400818258205d5f5c04aaa2367c5a700cf6ba9e9da76e214a0a1485a174618cb38b292bf0d9000182825839016a2fcce35ec3795b9418ae49b69074a17cdd0a7c60ae6ba63fc85eff17eabf85728a590b7785f27d60dea7d4bcb356b438b9d577a45547fe1b0000001e3001052482584c82d818584283581c91d0a0518e3e764e13f6ef37580a6be8ab14da4f3066fd01af01da6aa101581e581cabbf051bdee353839fbb21a6d4e6c584138a6a33896bb96d4124a330001a3592e2cc1a0a6526b0021a0002964d031a012f6296a10081825820e8fe69f9fd8afcb4792e3ca0f08b49e6eece1788c2d7b026096cfdbd1344a9bc5840dcef77b73af0922005f4b60d21333628348864c405ff52efd3f72523bf2c790e662650ad9951d306b40ce5beddf5b8eebb6731156b8b7617f6614b9ffdf2fb05f6""" tx = Transaction.from_cbor(tx_cbor_hex)