From f12c9b93b55f676204005ffe86c6d733dffd429b Mon Sep 17 00:00:00 2001 From: Jerry Date: Mon, 5 Oct 2026 20:22:25 -0700 Subject: [PATCH 1/4] Preserve CBOR container representation through decode Decoding normalises choices that are part of the encoding, so re-encoding a value produces different bytes and therefore a different hash. #503 fixed this for arrays; the same loss still applied to indefinite-length strings and maps, and the array fix had gaps of its own. Indefinite strings and maps now survive a round trip. The scanner already recorded that a container was indefinite and the apply step discarded it, so that flag is carried through to four new types. Chunk boundaries of an indefinite string are part of the encoding, so their lengths are recorded and replayed rather than renormalised. This is what the ledger needs: Plutus byte strings longer than 64 bytes arrive chunked, and such a datum previously re-encoded to a different hash (fixes #507). Also fixes three defects in the array path: - Duplicate map keys collapse on decode, leaving the shape walk misaligned with the decoded dict and silently producing wrong bytes. Such payloads are now rejected, and the opt-in path raises rather than guessing. - Trailing bytes and a non-COSE message body were guarded by bare asserts, which vanish under python -O. They now raise DeserializeException. The COSE guard could never fire as written, since verify() prepends the tag byte itself; it now checks the message body, as pycose did before #503. - A tag consumed by a semantic decoder left its content immutable, so the self-describe tag decoded to a tuple instead of a list. Type guards keep decoders that return a non-container (Fraction, Decimal, bignum, datetime) untouched. Scalar shapes are now shared rather than allocated per decoded item, which offsets most of the cost of the added work. Co-Authored-By: Claude Opus 5 (1M context) --- pycardano/cip/cip8.py | 11 +- pycardano/serialization.py | 261 +++++++++++++++++++++++++-- test/pycardano/test_cip8.py | 9 + test/pycardano/test_plutus.py | 18 ++ test/pycardano/test_serialization.py | 129 +++++++++++++ 5 files changed, 408 insertions(+), 20 deletions(-) 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..26f887b5 100644 --- a/pycardano/serialization.py +++ b/pycardano/serialization.py @@ -49,6 +49,10 @@ "default_encoder", "IndefiniteList", "IndefiniteFrozenList", + "IndefiniteByteString", + "IndefiniteTextString", + "IndefiniteDict", + "IndefiniteFrozenDict", "Primitive", "CBORBase", "CBORSerializable", @@ -82,6 +86,103 @@ class IndefiniteFrozenList(FrozenList, IndefiniteList): # type: ignore pass +class IndefiniteByteString(bytes): + """A byte string that was CBOR encoded as indefinite length chunks. + + Cardano splits byte strings longer than 64 bytes into 64 byte chunks, so the + chunk lengths are part of the encoding and therefore part of the hash. Decoding + records the original lengths so the value re-encodes to the bytes it came from. + """ + + chunks: tuple + + def __new__(cls, value: bytes, chunks: Optional[Iterable[int]] = None): + self = super().__new__(cls, value) + if chunks is None: + self.chunks = tuple( + min(64, len(value) - start) for start in range(0, len(value), 64) + ) + else: + self.chunks = tuple(chunks) + if sum(self.chunks) != len(value): + raise ValueError( + f"Chunk lengths {self.chunks} do not add up to the {len(value)} byte " + f"value they describe." + ) + return self + + 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(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. + """ + + chunks: tuple + + def __new__(cls, value: str, chunks: Optional[Iterable[int]] = None): + self = super().__new__(cls, value) + encoded = value.encode("utf-8") + if chunks is None: + self.chunks = (len(encoded),) if encoded else () + else: + self.chunks = tuple(chunks) + if sum(self.chunks) != len(encoded): + raise ValueError( + f"Chunk lengths {self.chunks} do not add up to the {len(encoded)} byte " + f"value they describe." + ) + return self + + 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(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] + + @dataclass class ByteString: value: bytes @@ -194,6 +295,23 @@ def wrapper(cls, value: Primitive): CBORBase = TypeVar("CBORBase", bound="CBORSerializable") +def _write_indefinite_chunks( + encoder: CBOREncoder, + header: bytes, + payload: bytes, + chunks: Iterable[int], + text: bool, +) -> None: + """Write one indefinite-length string as the exact chunks it was encoded with.""" + encoder.write(header) + start = 0 + for length in chunks: + chunk = payload[start : start + length] + encoder.encode(chunk.decode("utf-8") if text else chunk) + start += length + encoder.write(b"\xff") + + def default_encoder( encoder: CBOREncoder, value: Union[CBORSerializable, IndefiniteList] ): @@ -208,6 +326,9 @@ def default_encoder( FrozenList, IndefiniteFrozenList, FrozenDict, + IndefiniteByteString, + IndefiniteTextString, + IndefiniteDict, ), ), ( f"Type of input value is not CBORSerializable, " f"got {type(value)} instead." @@ -219,15 +340,37 @@ def default_encoder( for item in value: encoder.encode(item) encoder.write(b"\xff") + elif isinstance(value, IndefiniteByteString): + _write_indefinite_chunks( + encoder, b"\x5f", bytes(value), value.chunks, text=False + ) + elif isinstance(value, IndefiniteTextString): + _write_indefinite_chunks( + encoder, b"\x7f", str(value).encode("utf-8"), value.chunks, text=True + ) + 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_chunks( + encoder, b"\x5f", bytes(inner), inner.chunks, text=False + ) + elif len(inner) > 64: + _write_indefinite_chunks( + encoder, + b"\x5f", + inner, + (min(64, len(inner) - i) for i in range(0, len(inner), 64)), + text=False, + ) else: - encoder.encode(value.value) + encoder.encode(inner) elif isinstance(value, RawCBOR): encoder.write(value.cbor) elif isinstance(value, FrozenList): @@ -242,8 +385,17 @@ 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 ( + IndefiniteList, + IndefiniteFrozenList, + IndefiniteByteString, + IndefiniteTextString, + IndefiniteDict, + IndefiniteFrozenDict, + ): + # 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 +408,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 +440,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 +495,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 +524,21 @@ 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, most likely to duplicate keys, so the + # scanned shapes can no longer be matched to the values they came from. + # Guessing would silently re-encode the map to different bytes. + raise DeserializeException( + f"CBOR map was encoded with {len(shape.children)} entries but decoded " + f"to {len(value)}. Duplicate keys prevent PyCardano from restoring the " + f"original encoding; decode with allow_duplicate_keys=False to reject " + f"such payloads up front." + ) + paired_items = zip(value.items(), shape.children) mapped_items = { _apply_cbor_shape(key, key_shape, immutable=True): _apply_cbor_shape( @@ -344,11 +546,25 @@ 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 @@ -357,9 +573,16 @@ def loads(payload: bytes, **kwargs: Any) -> Any: """Decode CBOR while preserving order and indefinite-length arrays.""" 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..0dfc2bf3 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,9 @@ CBORSerializable, CodedSerializable, DictCBORSerializable, + IndefiniteByteString, IndefiniteList, + IndefiniteTextString, MapCBORSerializable, NonEmptyOrderedSet, OrderedSet, @@ -1171,6 +1175,131 @@ 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_still_refuse_to_guess(): + """Opting back into duplicate keys must not silently mis-apply array shapes.""" + encoded = bytes.fromhex("a2018102019f02ff") + + with pytest.raises(DeserializeException, match="Duplicate keys"): + loads(encoded, allow_duplicate_keys=True) + + +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) + + +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])) From daa0dff4b7a6a0c73a8fc22dd26be232d6b63171 Mon Sep 17 00:00:00 2001 From: Jerry Date: Mon, 5 Oct 2026 20:23:38 -0700 Subject: [PATCH 2/4] Pin Poetry in CI `pipx install poetry` resolved to whatever was newest when a job ran, so an unchanged commit could pass one day and fail the next. 2.4.2 was published two days before #502 and broke the integration tests that way. It also matters for the lock file. Poetry stamps poetry.lock with a content hash, and versions differ in how they compute it, so a mismatched Poetry rejects the lock and refuses to sync the environment. Pinning to the generator recorded in the lock file's first line keeps the two in agreement; the version is declared once per workflow so there is a single place to bump it. publish.yml is pinned too, since it runs `poetry install` and `poetry check` before pushing to PyPI. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/main.yml | 9 +++++++-- .github/workflows/publish.yml | 6 +++++- 2 files changed, 12 insertions(+), 3 deletions(-) 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: From 4dc78d65e5a523a643e18000dc901f29c3928bc7 Mon Sep 17 00:00:00 2001 From: Jerry Date: Mon, 5 Oct 2026 20:35:27 -0700 Subject: [PATCH 3/4] Cover the #402 signing failure for chunked datum bytestrings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The existing #402 fixture only holds byte strings short enough to fit one definite-length chunk, so it passed before this branch. Real DEX and escrow datums carry longer values, which the ledger requires to be chunked, and those still lost their chunk boundaries on decode: the body hash moved and a signature computed over the original body stopped verifying. The new test exercises that path end to end — sign the body hash as a wallet would before export, decode, then verify the same signature against the re-encoded body. It fails on 877be3a at the body hash assertion. Co-Authored-By: Claude Opus 5 (1M context) --- test/pycardano/test_transaction.py | 55 +++++++++++++++++++++++++++++- 1 file changed, 54 insertions(+), 1 deletion(-) 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) From 4698edcdbf9186d6565a5d4cdbf9bc23b4c9440d Mon Sep 17 00:00:00 2001 From: Jerry Date: Sat, 10 Oct 2026 12:56:37 -0700 Subject: [PATCH 4/4] Address review: shared Indefinite base, and keep duplicate-key maps Two changes from @nielstron's review. An `Indefinite` marker is now the common base of every indefinite type, and `IndefiniteString` holds the chunk bookkeeping that the byte and text variants share. `default_encoder` dispatches on those two bases instead of listing six concrete classes, its assertion lists one entry rather than four, and the two chunk-writing branches collapse into one shared path that the long-`ByteString` fallback reuses as well. Each concrete class is still registered with cbor2 individually, since it looks encoders up by exact type; that list is now a module-level tuple so the registration loop has a single source. Decoding with `allow_duplicate_keys=True` no longer raises. The caller asked for lenient decoding, so the decoded map is returned with its keys intact. What cannot be recovered is the representation: a map holds one entry per distinct key, so the scanned shapes can no longer be matched to the values they came from, and re-encoding will not reproduce the input bytes. That is the reason the default still rejects duplicates, and it is now stated on `loads` itself. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YMubCsu1qpLrc6iMt8eq9z --- pycardano/serialization.py | 209 ++++++++++++++++----------- test/pycardano/test_serialization.py | 30 +++- 2 files changed, 151 insertions(+), 88 deletions(-) diff --git a/pycardano/serialization.py b/pycardano/serialization.py index 26f887b5..82ae0c6c 100644 --- a/pycardano/serialization.py +++ b/pycardano/serialization.py @@ -47,8 +47,10 @@ __all__ = [ "default_encoder", + "Indefinite", "IndefiniteList", "IndefiniteFrozenList", + "IndefiniteString", "IndefiniteByteString", "IndefiniteTextString", "IndefiniteDict", @@ -74,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 @@ -86,31 +97,73 @@ class IndefiniteFrozenList(FrozenList, IndefiniteList): # type: ignore pass -class IndefiniteByteString(bytes): - """A byte string that was CBOR encoded as indefinite length chunks. +class IndefiniteString(Indefinite): + """Shared chunk bookkeeping for the two indefinite-length string types. - Cardano splits byte strings longer than 64 bytes into 64 byte chunks, so the - chunk lengths are part of the encoding and therefore part of the hash. Decoding - records the original lengths so the value re-encodes to the bytes it came from. + 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 - def __new__(cls, value: bytes, chunks: Optional[Iterable[int]] = None): - self = super().__new__(cls, value) - if chunks is None: - self.chunks = tuple( - min(64, len(value) - start) for start in range(0, len(value), 64) - ) - else: - self.chunks = tuple(chunks) - if sum(self.chunks) != len(value): + @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 {self.chunks} do not add up to the {len(value)} byte " + 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})" @@ -121,28 +174,30 @@ def __reduce__(self): return (IndefiniteByteString, (bytes(self), self.chunks)) -class IndefiniteTextString(str): +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. """ - chunks: tuple + CBOR_HEADER = b"\x7f" def __new__(cls, value: str, chunks: Optional[Iterable[int]] = None): self = super().__new__(cls, value) - encoded = value.encode("utf-8") - if chunks is None: - self.chunks = (len(encoded),) if encoded else () - else: - self.chunks = tuple(chunks) - if sum(self.chunks) != len(encoded): - raise ValueError( - f"Chunk lengths {self.chunks} do not add up to the {len(encoded)} byte " - f"value they describe." - ) + 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})" @@ -153,7 +208,7 @@ def __reduce__(self): return (IndefiniteTextString, (str(self), self.chunks)) -class IndefiniteDict(dict): +class IndefiniteDict(Indefinite, dict): """A map that was CBOR encoded with an indefinite length header.""" def __repr__(self): @@ -183,6 +238,18 @@ def _immutable(self, *args, **kwargs): 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 @@ -295,20 +362,11 @@ def wrapper(cls, value: Primitive): CBORBase = TypeVar("CBORBase", bound="CBORSerializable") -def _write_indefinite_chunks( - encoder: CBOREncoder, - header: bytes, - payload: bytes, - chunks: Iterable[int], - text: bool, -) -> None: - """Write one indefinite-length string as the exact chunks it was encoded with.""" - encoder.write(header) - start = 0 - for length in chunks: - chunk = payload[start : start + length] - encoder.encode(chunk.decode("utf-8") if text else chunk) - start += length +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") @@ -321,14 +379,10 @@ def default_encoder( ( ByteString, CBORSerializable, - IndefiniteList, + Indefinite, RawCBOR, FrozenList, - IndefiniteFrozenList, FrozenDict, - IndefiniteByteString, - IndefiniteTextString, - IndefiniteDict, ), ), ( f"Type of input value is not CBORSerializable, " f"got {type(value)} instead." @@ -340,14 +394,8 @@ def default_encoder( for item in value: encoder.encode(item) encoder.write(b"\xff") - elif isinstance(value, IndefiniteByteString): - _write_indefinite_chunks( - encoder, b"\x5f", bytes(value), value.chunks, text=False - ) - elif isinstance(value, IndefiniteTextString): - _write_indefinite_chunks( - encoder, b"\x7f", str(value).encode("utf-8"), value.chunks, text=True - ) + 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(): @@ -358,17 +406,11 @@ def default_encoder( inner = value.value if isinstance(inner, IndefiniteByteString): # The wrapped value remembers how it arrived, so reuse those chunks. - _write_indefinite_chunks( - encoder, b"\x5f", bytes(inner), inner.chunks, text=False - ) + _write_indefinite_string(encoder, inner) elif len(inner) > 64: - _write_indefinite_chunks( - encoder, - b"\x5f", - inner, - (min(64, len(inner) - i) for i in range(0, len(inner), 64)), - text=False, - ) + # 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(inner) elif isinstance(value, RawCBOR): @@ -385,14 +427,7 @@ 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) - for indefinite_type in ( - IndefiniteList, - IndefiniteFrozenList, - IndefiniteByteString, - IndefiniteTextString, - IndefiniteDict, - IndefiniteFrozenDict, - ): + 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) @@ -529,15 +564,14 @@ def _apply_cbor_shape(value: Any, shape: _CBORShape, immutable: bool = False) -> return value if len(value) != len(shape.children): - # Entries were lost on the way in, most likely to duplicate keys, so the - # scanned shapes can no longer be matched to the values they came from. - # Guessing would silently re-encode the map to different bytes. - raise DeserializeException( - f"CBOR map was encoded with {len(shape.children)} entries but decoded " - f"to {len(value)}. Duplicate keys prevent PyCardano from restoring the " - f"original encoding; decode with allow_duplicate_keys=False to reject " - f"such payloads up front." - ) + # 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 = { @@ -570,7 +604,14 @@ def _apply_cbor_shape(value: Any, shape: _CBORShape, immutable: bool = False) -> 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 diff --git a/test/pycardano/test_serialization.py b/test/pycardano/test_serialization.py index 0dfc2bf3..bd5f7350 100644 --- a/test/pycardano/test_serialization.py +++ b/test/pycardano/test_serialization.py @@ -50,6 +50,7 @@ CBORSerializable, CodedSerializable, DictCBORSerializable, + Indefinite, IndefiniteByteString, IndefiniteList, IndefiniteTextString, @@ -1184,12 +1185,19 @@ def test_duplicate_map_keys_are_rejected(): loads(encoded) -def test_duplicate_map_keys_allowed_explicitly_still_refuse_to_guess(): - """Opting back into duplicate keys must not silently mis-apply array shapes.""" +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") - with pytest.raises(DeserializeException, match="Duplicate keys"): - loads(encoded, allow_duplicate_keys=True) + decoded = loads(encoded, allow_duplicate_keys=True) + + assert decoded == {1: [2]} + assert dumps(decoded) == bytes.fromhex("a1018102") def test_trailing_bytes_raise_deserialize_exception(): @@ -1269,6 +1277,20 @@ def test_indefinite_byte_string_behaves_like_bytes(): 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))