Streamable.from_bytes() (#20682)

* Streamable.from_bytes() is currently failing with an assert when not all bytes are consumed. Promote it to a proper run-time error (it's not a programmer error)

* fixup LegacyCATInfo

chia/_tests/core/util/test_streamable.py

@@ -780,10 +780,25 @@ def test_ambiguous_deserialization_program() -> None:
 
     TestClassProgram.from_bytes(bytes(program))
 
-    with pytest.raises(AssertionError):
+    with pytest.raises(ValueError):
         TestClassProgram.from_bytes(bytes(program) + b"9")
 
 
+def test_from_bytes_rejects_trailing_bytes_rust_types() -> None:
+    from chia_rs import G2Element, SpendBundle
+
+    coin = Coin(bytes32(bytes(32)), bytes32(bytes(32)), uint64(0))
+    valid_coin = bytes(coin)
+    Coin.from_bytes(valid_coin)
+    with pytest.raises(ValueError):
+        Coin.from_bytes(valid_coin + b"\x00")
+
+    valid_sb = bytes(4) + bytes(G2Element())
+    SpendBundle.from_bytes(valid_sb)
+    with pytest.raises(ValueError):
+        SpendBundle.from_bytes(valid_sb + b"\x00")
+
+
 def test_streamable_empty() -> None:
     @streamable
     @dataclass(frozen=True)

chia/util/streamable.py

@@ -683,7 +683,9 @@ class Streamable:
     def from_bytes(cls, blob: bytes) -> Self:
         f = io.BytesIO(blob)
         parsed = cls.parse(f)
-        assert f.read() == b""
+        remainder = f.read()
+        if remainder != b"":
+            raise ValueError(f"{cls.__name__}: {len(remainder)} bytes not consumed")
         return parsed
 
     def stream_to_bytes(self) -> bytes:

chia/wallet/cat_wallet/cat_wallet.py

@@ -292,7 +292,7 @@ class CATWallet:
         try:
             self.cat_info = cls.wallet_info_type.from_bytes(hexstr_to_bytes(self.wallet_info.data))
             self.lineage_store = await CATLineageStore.create(self.wallet_state_manager.db_wrapper, self.get_asset_id())
-        except AssertionError:
+        except (AssertionError, ValueError):
             # Do a migration of the lineage proofs
             cat_info = LegacyCATInfo.from_bytes(hexstr_to_bytes(self.wallet_info.data))
             self.cat_info = cls.wallet_info_type(cat_info.limitations_program_hash, cat_info.my_tail)