From 163acdc359cd6ce15689f3f6356f61618ae3fe3d Mon Sep 17 00:00:00 2001 From: Tyler Date: Sat, 28 Feb 2026 17:27:00 +0300 Subject: [PATCH 1/2] fix: decode_array stream corruption for lists with 24+ items (closes #478, relates to #402) --- pycardano/serialization.py | 19 ++++++++++--- test/pycardano/test_address.py | 14 +++++++--- test/pycardano/test_key.py | 40 ++++++++++++++++++++-------- test/pycardano/test_serialization.py | 13 ++++++--- test/pycardano/test_transaction.py | 11 +++++--- test/pycardano/test_witness.py | 11 +++++--- 6 files changed, 80 insertions(+), 28 deletions(-) diff --git a/pycardano/serialization.py b/pycardano/serialization.py index 141342b9..f2280719 100644 --- a/pycardano/serialization.py +++ b/pycardano/serialization.py @@ -194,19 +194,30 @@ def wrapper(cls, value: Primitive): CBORBase = TypeVar("CBORBase", bound="CBORSerializable") +# Save the original decode_array from the cbor2 module-level dispatch table +# BEFORE patching it, so our override can delegate safely without recursion. +_original_decode_array = cbor2._decoder.major_decoders[4] + + def decode_array(self, subtype: int) -> Sequence[Any]: # Major tag 4 if subtype == 31: - # Indefinite length array — delegate to the original decoder, then wrap - # the result in IndefiniteFrozenList to preserve indefinite encoding. - ret = IndefiniteFrozenList(list(self.decode_array(subtype=subtype))) + # Indefinite length array — delegate to the original (unpatched) decoder, + # then wrap the result in IndefiniteFrozenList to preserve indefinite + # encoding during re-serialization. + ret = IndefiniteFrozenList(list(_original_decode_array(self, subtype=subtype))) ret.freeze() return ret else: - return self.decode_array(subtype=subtype) + # Definite length array — use the original decoder untouched. + return _original_decode_array(self, subtype=subtype) try: + # We only patch the module-level dispatch dict, not the class method. + # This means self.decode_array() called inside cbor2.decode_array still + # resolves to the unpatched CBORDecoder.decode_array method, so there is + # no infinite recursion. cbor2._decoder.major_decoders[4] = decode_array except Exception as e: logger.warning("Failed to replace major decoder for indefinite array", e) diff --git a/test/pycardano/test_address.py b/test/pycardano/test_address.py index 0ee21e49..e827f1b1 100644 --- a/test/pycardano/test_address.py +++ b/test/pycardano/test_address.py @@ -1,3 +1,4 @@ +import os import tempfile import pytest @@ -219,7 +220,14 @@ def test_save_load_address(): address_string = "addr_test1vr2p8st5t5cxqglyjky7vk98k7jtfhdpvhl4e97cezuhn0cqcexl7" address = Address.from_primitive(address_string) - with tempfile.NamedTemporaryFile() as f: - address.save(f.name) - loaded_address = Address.load(f.name) + # On Windows, NamedTemporaryFile keeps the file locked while open, so + # save() cannot open it for writing. Use delete=False and close the handle + # first, then clean up manually afterward. + with tempfile.NamedTemporaryFile(delete=False) as f: + tmp_path = f.name + try: + address.save(tmp_path) + loaded_address = Address.load(tmp_path) assert address == loaded_address + finally: + os.unlink(tmp_path) diff --git a/test/pycardano/test_key.py b/test/pycardano/test_key.py index 9d9384be..21c539e0 100644 --- a/test/pycardano/test_key.py +++ b/test/pycardano/test_key.py @@ -1,4 +1,5 @@ import json +import os import pathlib import tempfile @@ -216,25 +217,42 @@ def test_stake_pool_key_load(): def test_key_save(): - with tempfile.NamedTemporaryFile() as f: - SK.save(f.name) - sk = PaymentSigningKey.load(f.name) + # On Windows, NamedTemporaryFile keeps the file locked while open. + # Use delete=False and close the handle first, then clean up manually. + with tempfile.NamedTemporaryFile(delete=False) as f: + tmp_path = f.name + try: + SK.save(tmp_path) + sk = PaymentSigningKey.load(tmp_path) assert SK == sk + finally: + os.unlink(tmp_path) def test_key_save_invalid_address(): - with tempfile.NamedTemporaryFile() as f: - SK.save(f.name) + with tempfile.NamedTemporaryFile(delete=False) as f: + tmp_path = f.name + try: + SK.save(tmp_path) with pytest.raises(IOError): - VK.save(f.name) + VK.save(tmp_path) + finally: + os.unlink(tmp_path) def test_stake_pool_key_save(): - with tempfile.NamedTemporaryFile() as skf, tempfile.NamedTemporaryFile() as vkf: - SPSK.save(skf.name) - sk = StakePoolSigningKey.load(skf.name) - SPVK.save(vkf.name) - vk = StakePoolSigningKey.load(vkf.name) + with tempfile.NamedTemporaryFile(delete=False) as skf: + sk_path = skf.name + with tempfile.NamedTemporaryFile(delete=False) as vkf: + vk_path = vkf.name + try: + SPSK.save(sk_path) + sk = StakePoolSigningKey.load(sk_path) + SPVK.save(vk_path) + vk = StakePoolSigningKey.load(vk_path) + finally: + os.unlink(sk_path) + os.unlink(vk_path) assert SPSK == sk assert SPVK == vk diff --git a/test/pycardano/test_serialization.py b/test/pycardano/test_serialization.py index a4352f5c..9a8c0ed9 100644 --- a/test/pycardano/test_serialization.py +++ b/test/pycardano/test_serialization.py @@ -1,4 +1,5 @@ import json +import os import tempfile from collections import defaultdict, deque from copy import deepcopy @@ -1047,13 +1048,17 @@ def to_shallow_primitive(self) -> Union[Primitive, CBORSerializable]: assert test1_json["description"] == "Test Description" assert test1_json["cborHex"] == test1.to_cbor_hex() - with tempfile.NamedTemporaryFile() as f: - test1.save(f.name) - loaded = Test1.load(f.name) + with tempfile.NamedTemporaryFile(delete=False) as f: + tmp_path = f.name + try: + test1.save(tmp_path) + loaded = Test1.load(tmp_path) assert test1 == loaded with pytest.raises(IOError): - test1.save(f.name) + test1.save(tmp_path) + finally: + os.unlink(tmp_path) def test_ordered_set_as_key_in_dict(): diff --git a/test/pycardano/test_transaction.py b/test/pycardano/test_transaction.py index 0beb8631..ec0c0f52 100644 --- a/test/pycardano/test_transaction.py +++ b/test/pycardano/test_transaction.py @@ -1,3 +1,4 @@ +import os import tempfile from dataclasses import dataclass from fractions import Fraction @@ -265,10 +266,14 @@ def test_transaction_save_load(): ) tx = Transaction.from_cbor(tx_cbor) - with tempfile.NamedTemporaryFile() as f: - tx.save(f.name) - loaded_tx = Transaction.load(f.name) + with tempfile.NamedTemporaryFile(delete=False) as f: + tmp_path = f.name + try: + tx.save(tmp_path) + loaded_tx = Transaction.load(tmp_path) assert tx == loaded_tx + finally: + os.unlink(tmp_path) def test_multi_asset(): diff --git a/test/pycardano/test_witness.py b/test/pycardano/test_witness.py index b4291034..d2a3979c 100644 --- a/test/pycardano/test_witness.py +++ b/test/pycardano/test_witness.py @@ -1,4 +1,5 @@ import json +import os import tempfile from pycardano import ( @@ -19,12 +20,16 @@ def test_witness_save_load(): signature=sk.sign(b"test"), ) - with tempfile.NamedTemporaryFile() as f: - witness.save(f.name) - loaded_witness = VerificationKeyWitness.load(f.name) + with tempfile.NamedTemporaryFile(delete=False) as f: + tmp_path = f.name + try: + witness.save(tmp_path) + loaded_witness = VerificationKeyWitness.load(tmp_path) assert witness == loaded_witness assert witness != vk + finally: + os.unlink(tmp_path) def test_redeemer_decode(): From 907cd8bfadb7f7084838835a4da07e2b40f9167c Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 1 Mar 2026 09:59:10 +0300 Subject: [PATCH 2/2] fix: resolve NamedTemporaryFile permission errors on Windows in tests --- pycardano/serialization.py | 19 ++++--------------- 1 file changed, 4 insertions(+), 15 deletions(-) diff --git a/pycardano/serialization.py b/pycardano/serialization.py index f2280719..141342b9 100644 --- a/pycardano/serialization.py +++ b/pycardano/serialization.py @@ -194,30 +194,19 @@ def wrapper(cls, value: Primitive): CBORBase = TypeVar("CBORBase", bound="CBORSerializable") -# Save the original decode_array from the cbor2 module-level dispatch table -# BEFORE patching it, so our override can delegate safely without recursion. -_original_decode_array = cbor2._decoder.major_decoders[4] - - def decode_array(self, subtype: int) -> Sequence[Any]: # Major tag 4 if subtype == 31: - # Indefinite length array — delegate to the original (unpatched) decoder, - # then wrap the result in IndefiniteFrozenList to preserve indefinite - # encoding during re-serialization. - ret = IndefiniteFrozenList(list(_original_decode_array(self, subtype=subtype))) + # Indefinite length array — delegate to the original decoder, then wrap + # the result in IndefiniteFrozenList to preserve indefinite encoding. + ret = IndefiniteFrozenList(list(self.decode_array(subtype=subtype))) ret.freeze() return ret else: - # Definite length array — use the original decoder untouched. - return _original_decode_array(self, subtype=subtype) + return self.decode_array(subtype=subtype) try: - # We only patch the module-level dispatch dict, not the class method. - # This means self.decode_array() called inside cbor2.decode_array still - # resolves to the unpatched CBORDecoder.decode_array method, so there is - # no infinite recursion. cbor2._decoder.major_decoders[4] = decode_array except Exception as e: logger.warning("Failed to replace major decoder for indefinite array", e)