From 14cdae6c0813fe6a381070e0c9f14c217e049b17 Mon Sep 17 00:00:00 2001 From: unnawut Date: Tue, 10 Feb 2026 14:28:30 +0700 Subject: [PATCH 1/3] add ssz test vectors --- .../testing/src/consensus_testing/__init__.py | 4 + .../test_fixtures/__init__.py | 2 + .../consensus_testing/test_fixtures/ssz.py | 86 ++++ src/lean_spec/subspecs/ssz/hash.py | 15 +- tests/consensus/devnet/ssz/__init__.py | 1 + .../devnet/ssz/test_consensus_containers.py | 436 ++++++++++++++++++ .../devnet/ssz/test_networking_containers.py | 70 +++ .../devnet/ssz/test_xmss_containers.py | 230 +++++++++ 8 files changed, 840 insertions(+), 4 deletions(-) create mode 100644 packages/testing/src/consensus_testing/test_fixtures/ssz.py create mode 100644 tests/consensus/devnet/ssz/__init__.py create mode 100644 tests/consensus/devnet/ssz/test_consensus_containers.py create mode 100644 tests/consensus/devnet/ssz/test_networking_containers.py create mode 100644 tests/consensus/devnet/ssz/test_xmss_containers.py diff --git a/packages/testing/src/consensus_testing/__init__.py b/packages/testing/src/consensus_testing/__init__.py index 70b983c5..875be894 100644 --- a/packages/testing/src/consensus_testing/__init__.py +++ b/packages/testing/src/consensus_testing/__init__.py @@ -6,6 +6,7 @@ from .test_fixtures import ( BaseConsensusFixture, ForkChoiceTest, + SSZTest, StateTransitionTest, VerifySignaturesTest, ) @@ -27,6 +28,7 @@ StateTransitionTestFiller = Type[StateTransitionTest] ForkChoiceTestFiller = Type[ForkChoiceTest] VerifySignaturesTestFiller = Type[VerifySignaturesTest] +SSZTestFiller = Type[SSZTest] __all__ = [ # Public API @@ -40,6 +42,7 @@ "StateTransitionTest", "ForkChoiceTest", "VerifySignaturesTest", + "SSZTest", # Test types "BaseForkChoiceStep", "TickStep", @@ -54,4 +57,5 @@ "StateTransitionTestFiller", "ForkChoiceTestFiller", "VerifySignaturesTestFiller", + "SSZTestFiller", ] diff --git a/packages/testing/src/consensus_testing/test_fixtures/__init__.py b/packages/testing/src/consensus_testing/test_fixtures/__init__.py index 821b23b5..6f26796f 100644 --- a/packages/testing/src/consensus_testing/test_fixtures/__init__.py +++ b/packages/testing/src/consensus_testing/test_fixtures/__init__.py @@ -2,6 +2,7 @@ from .base import BaseConsensusFixture from .fork_choice import ForkChoiceTest +from .ssz import SSZTest from .state_transition import StateTransitionTest from .verify_signatures import VerifySignaturesTest @@ -10,4 +11,5 @@ "StateTransitionTest", "ForkChoiceTest", "VerifySignaturesTest", + "SSZTest", ] diff --git a/packages/testing/src/consensus_testing/test_fixtures/ssz.py b/packages/testing/src/consensus_testing/test_fixtures/ssz.py new file mode 100644 index 00000000..08cfb763 --- /dev/null +++ b/packages/testing/src/consensus_testing/test_fixtures/ssz.py @@ -0,0 +1,86 @@ +"""SSZ test fixture format for serialization conformance testing.""" + +from typing import Any, ClassVar + +from pydantic import field_serializer + +from lean_spec.subspecs.ssz.hash import hash_tree_root +from lean_spec.types import Bytes32 +from lean_spec.types.container import Container + +from .base import BaseConsensusFixture + + +class SSZTest(BaseConsensusFixture): + """ + Test fixture for SSZ serialization/deserialization conformance. + + Tests roundtrip serialization and hash tree root computation for SSZ containers. + + Structure: + type_name: Name of the container class + value: The container instance + serialized: Hex-encoded SSZ bytes (computed) + root: Hex-encoded hash tree root (computed) + """ + + format_name: ClassVar[str] = "ssz" + description: ClassVar[str] = "Tests SSZ serialization roundtrip and hash tree root computation" + + type_name: str + """Name of the container class being tested.""" + + value: Container + """The container instance to test.""" + + serialized: str = "" + """Hex-encoded SSZ serialized bytes (computed during make_fixture).""" + + root: str = "" + """Hex-encoded hash tree root (computed during make_fixture).""" + + @field_serializer("value", when_used="json") + def serialize_value(self, value: Container) -> dict[str, Any]: + """Serialize the container value to JSON using its native serialization.""" + return value.to_json() + + def make_fixture(self) -> "SSZTest": + """ + Generate the fixture by testing SSZ roundtrip and computing root. + + 1. Serialize the value to SSZ bytes + 2. Deserialize the bytes back to a container + 3. Verify the roundtrip produces the same value + 4. Compute the hash tree root + + Returns: + SSZTest with computed serialized and root fields. + + Raises: + AssertionError: If roundtrip fails or types don't match. + """ + # Serialize to SSZ bytes + ssz_bytes = self.value.encode_bytes() + + # Deserialize back + container_cls = type(self.value) + decoded = container_cls.decode_bytes(ssz_bytes) + + # Verify roundtrip + assert decoded == self.value, ( + f"SSZ roundtrip failed for {self.type_name}: " + f"original != decoded\n" + f"Original: {self.value}\n" + f"Decoded: {decoded}" + ) + + # Compute hash tree root + htr: Bytes32 = hash_tree_root(self.value) + + # Return fixture with computed fields + return self.model_copy( + update={ + "serialized": "0x" + ssz_bytes.hex(), + "root": "0x" + htr.hex(), + } + ) diff --git a/src/lean_spec/subspecs/ssz/hash.py b/src/lean_spec/subspecs/ssz/hash.py index febdeaac..b4baaac8 100644 --- a/src/lean_spec/subspecs/ssz/hash.py +++ b/src/lean_spec/subspecs/ssz/hash.py @@ -14,6 +14,7 @@ from functools import singledispatch from math import ceil +from lean_spec.subspecs.koalabear import Fp from lean_spec.subspecs.ssz.constants import BITS_PER_CHUNK, BYTES_PER_CHUNK from lean_spec.types.bitfields import BaseBitlist, BaseBitvector from lean_spec.types.boolean import Boolean @@ -53,6 +54,12 @@ def _htr_boolean(value: Boolean) -> Bytes32: return merkleize(pack_bytes(value.encode_bytes())) +@hash_tree_root.register +def _htr_fp(value: Fp) -> Bytes32: + """KoalaBear field elements: pack bytes into chunks and merkleize.""" + return merkleize(pack_bytes(value.encode_bytes())) + + @hash_tree_root.register def _htr_bytes(value: bytes) -> Bytes32: """Treat raw bytes like ByteVector[N].""" @@ -104,9 +111,9 @@ def _htr_bitlist_base(value: BaseBitlist) -> Bytes32: def _htr_vector(value: SSZVector) -> Bytes32: elem_t, length = type(value).ELEMENT_TYPE, type(value).LENGTH - if issubclass(elem_t, (BaseUint, Boolean)): + if issubclass(elem_t, (BaseUint, Boolean, Fp)): # BASIC elements: pack serialized bytes - elem_size = elem_t.get_byte_length() if issubclass(elem_t, BaseUint) else 1 + elem_size = elem_t.get_byte_length() # Compute limit in chunks: ceil((length * elem_size) / BYTES_PER_CHUNK) limit_chunks = (length * elem_size + BYTES_PER_CHUNK - 1) // BYTES_PER_CHUNK return merkleize( @@ -122,9 +129,9 @@ def _htr_vector(value: SSZVector) -> Bytes32: def _htr_list(value: SSZList) -> Bytes32: elem_t, limit = type(value).ELEMENT_TYPE, type(value).LIMIT - if issubclass(elem_t, (BaseUint, Boolean)): + if issubclass(elem_t, (BaseUint, Boolean, Fp)): # BASIC elements: pack serialized bytes - elem_size = elem_t.get_byte_length() if issubclass(elem_t, BaseUint) else 1 + elem_size = elem_t.get_byte_length() # Compute limit in chunks: ceil((limit * elem_size) / BYTES_PER_CHUNK) limit_chunks = (limit * elem_size + BYTES_PER_CHUNK - 1) // BYTES_PER_CHUNK root = merkleize( diff --git a/tests/consensus/devnet/ssz/__init__.py b/tests/consensus/devnet/ssz/__init__.py new file mode 100644 index 00000000..8cf28364 --- /dev/null +++ b/tests/consensus/devnet/ssz/__init__.py @@ -0,0 +1 @@ +"""SSZ serialization conformance tests.""" diff --git a/tests/consensus/devnet/ssz/test_consensus_containers.py b/tests/consensus/devnet/ssz/test_consensus_containers.py new file mode 100644 index 00000000..40f301ad --- /dev/null +++ b/tests/consensus/devnet/ssz/test_consensus_containers.py @@ -0,0 +1,436 @@ +"""SSZ conformance tests for consensus containers.""" + +import pytest +from consensus_testing import SSZTestFiller + +from lean_spec.subspecs.containers import ( + AggregatedAttestation, + Attestation, + AttestationData, + Block, + BlockBody, + BlockHeader, + BlockWithAttestation, + Checkpoint, + Config, + SignedAttestation, + SignedBlockWithAttestation, + Slot, + State, + Validator, + ValidatorIndex, +) +from lean_spec.subspecs.containers.attestation import AggregationBits +from lean_spec.subspecs.containers.block import BlockSignatures +from lean_spec.subspecs.containers.block.types import ( + AggregatedAttestations, + AttestationSignatures, +) +from lean_spec.subspecs.containers.state.types import ( + HistoricalBlockHashes, + JustificationRoots, + JustificationValidators, + JustifiedSlots, + Validators, +) +from lean_spec.subspecs.koalabear import Fp +from lean_spec.subspecs.xmss import Signature +from lean_spec.subspecs.xmss.aggregation import AggregatedSignatureProof +from lean_spec.subspecs.xmss.types import ( + HashDigestList, + HashTreeOpening, + Randomness, +) +from lean_spec.types import Boolean, Bytes32, Bytes52, Uint64 +from lean_spec.types.byte_arrays import ByteListMiB + +pytestmark = pytest.mark.valid_until("Devnet") + + +# --- Helper functions --- + + +def _zero_checkpoint() -> Checkpoint: + return Checkpoint(root=Bytes32.zero(), slot=Slot(0)) + + +def _zero_attestation_data() -> AttestationData: + zero_cp = _zero_checkpoint() + return AttestationData(slot=Slot(0), head=zero_cp, target=zero_cp, source=zero_cp) + + +def _typical_attestation_data() -> AttestationData: + head = Checkpoint(root=Bytes32(b"\x01" * 32), slot=Slot(100)) + target = Checkpoint(root=Bytes32(b"\x02" * 32), slot=Slot(99)) + source = Checkpoint(root=Bytes32(b"\x03" * 32), slot=Slot(50)) + return AttestationData(slot=Slot(100), head=head, target=target, source=source) + + +def _zero_signature() -> Signature: + return Signature( + path=HashTreeOpening(siblings=HashDigestList(data=[])), + rho=Randomness(data=[Fp(0) for _ in range(Randomness.LENGTH)]), + hashes=HashDigestList(data=[]), + ) + + +# --- Checkpoint --- + + +def test_checkpoint_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Checkpoint with zero values.""" + ssz(type_name="Checkpoint", value=_zero_checkpoint()) + + +def test_checkpoint_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Checkpoint with typical values.""" + ssz( + type_name="Checkpoint", + value=Checkpoint(root=Bytes32(b"\xab" * 32), slot=Slot(12345)), + ) + + +# --- AttestationData --- + + +def test_attestation_data_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for AttestationData with zero values.""" + ssz(type_name="AttestationData", value=_zero_attestation_data()) + + +def test_attestation_data_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for AttestationData with typical values.""" + ssz(type_name="AttestationData", value=_typical_attestation_data()) + + +# --- Attestation --- + + +def test_attestation_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Attestation with zero values.""" + ssz( + type_name="Attestation", + value=Attestation(validator_id=ValidatorIndex(0), data=_zero_attestation_data()), + ) + + +def test_attestation_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Attestation with typical values.""" + ssz( + type_name="Attestation", + value=Attestation(validator_id=ValidatorIndex(42), data=_typical_attestation_data()), + ) + + +# --- SignedAttestation --- + + +def test_signed_attestation_minimal(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for SignedAttestation with minimal values.""" + ssz( + type_name="SignedAttestation", + value=SignedAttestation( + validator_id=ValidatorIndex(0), + message=_zero_attestation_data(), + signature=_zero_signature(), + ), + ) + + +# --- AggregatedAttestation --- + + +def test_aggregated_attestation_single(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for AggregatedAttestation with single validator.""" + ssz( + type_name="AggregatedAttestation", + value=AggregatedAttestation( + aggregation_bits=AggregationBits(data=[Boolean(True)]), + data=_zero_attestation_data(), + ), + ) + + +def test_aggregated_attestation_multiple(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for AggregatedAttestation with multiple validators.""" + ssz( + type_name="AggregatedAttestation", + value=AggregatedAttestation( + aggregation_bits=AggregationBits( + data=[Boolean(True), Boolean(False), Boolean(True), Boolean(True)] + ), + data=_typical_attestation_data(), + ), + ) + + +# --- BlockBody --- + + +def test_block_body_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockBody with no attestations.""" + ssz( + type_name="BlockBody", + value=BlockBody(attestations=AggregatedAttestations(data=[])), + ) + + +def test_block_body_with_attestation(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockBody with attestations.""" + ssz( + type_name="BlockBody", + value=BlockBody( + attestations=AggregatedAttestations( + data=[ + AggregatedAttestation( + aggregation_bits=AggregationBits(data=[Boolean(True)]), + data=_zero_attestation_data(), + ) + ] + ) + ), + ) + + +# --- BlockHeader --- + + +def test_block_header_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockHeader with zero values.""" + ssz( + type_name="BlockHeader", + value=BlockHeader( + slot=Slot(0), + proposer_index=ValidatorIndex(0), + parent_root=Bytes32.zero(), + state_root=Bytes32.zero(), + body_root=Bytes32.zero(), + ), + ) + + +def test_block_header_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockHeader with typical values.""" + ssz( + type_name="BlockHeader", + value=BlockHeader( + slot=Slot(100), + proposer_index=ValidatorIndex(3), + parent_root=Bytes32(b"\x01" * 32), + state_root=Bytes32(b"\x02" * 32), + body_root=Bytes32(b"\x03" * 32), + ), + ) + + +# --- Block --- + + +def test_block_empty_body(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Block with empty body.""" + ssz( + type_name="Block", + value=Block( + slot=Slot(0), + proposer_index=ValidatorIndex(0), + parent_root=Bytes32.zero(), + state_root=Bytes32.zero(), + body=BlockBody(attestations=AggregatedAttestations(data=[])), + ), + ) + + +def test_block_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Block with attestations.""" + ssz( + type_name="Block", + value=Block( + slot=Slot(100), + proposer_index=ValidatorIndex(3), + parent_root=Bytes32(b"\x01" * 32), + state_root=Bytes32(b"\x02" * 32), + body=BlockBody( + attestations=AggregatedAttestations( + data=[ + AggregatedAttestation( + aggregation_bits=AggregationBits(data=[Boolean(True)]), + data=_typical_attestation_data(), + ) + ] + ) + ), + ), + ) + + +# --- BlockWithAttestation --- + + +def test_block_with_attestation_minimal(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockWithAttestation with minimal values.""" + block = Block( + slot=Slot(1), + proposer_index=ValidatorIndex(0), + parent_root=Bytes32.zero(), + state_root=Bytes32.zero(), + body=BlockBody(attestations=AggregatedAttestations(data=[])), + ) + attestation = Attestation(validator_id=ValidatorIndex(0), data=_zero_attestation_data()) + ssz( + type_name="BlockWithAttestation", + value=BlockWithAttestation(block=block, proposer_attestation=attestation), + ) + + +# --- BlockSignatures --- + + +def test_block_signatures_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockSignatures with no attestation signatures.""" + ssz( + type_name="BlockSignatures", + value=BlockSignatures( + attestation_signatures=AttestationSignatures(data=[]), + proposer_signature=_zero_signature(), + ), + ) + + +def test_block_signatures_with_attestation(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlockSignatures with attestation signatures.""" + ssz( + type_name="BlockSignatures", + value=BlockSignatures( + attestation_signatures=AttestationSignatures( + data=[ + AggregatedSignatureProof( + participants=AggregationBits(data=[Boolean(True)]), + proof_data=ByteListMiB(data=b""), + ) + ] + ), + proposer_signature=_zero_signature(), + ), + ) + + +# --- SignedBlockWithAttestation --- + + +def test_signed_block_with_attestation_minimal(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for SignedBlockWithAttestation with minimal values.""" + block = Block( + slot=Slot(1), + proposer_index=ValidatorIndex(0), + parent_root=Bytes32.zero(), + state_root=Bytes32.zero(), + body=BlockBody(attestations=AggregatedAttestations(data=[])), + ) + attestation = Attestation(validator_id=ValidatorIndex(0), data=_zero_attestation_data()) + message = BlockWithAttestation(block=block, proposer_attestation=attestation) + signature = BlockSignatures( + attestation_signatures=AttestationSignatures(data=[]), + proposer_signature=_zero_signature(), + ) + ssz( + type_name="SignedBlockWithAttestation", + value=SignedBlockWithAttestation(message=message, signature=signature), + ) + + +# --- Config --- + + +def test_config_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Config with zero genesis time.""" + ssz(type_name="Config", value=Config(genesis_time=Uint64(0))) + + +def test_config_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Config with typical genesis time.""" + ssz(type_name="Config", value=Config(genesis_time=Uint64(1609459200))) + + +# --- Validator --- + + +def test_validator_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Validator with zero values.""" + ssz( + type_name="Validator", + value=Validator(pubkey=Bytes52.zero(), index=ValidatorIndex(0)), + ) + + +def test_validator_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Validator with typical values.""" + ssz( + type_name="Validator", + value=Validator(pubkey=Bytes52(b"\xab" * 52), index=ValidatorIndex(42)), + ) + + +# --- State --- + + +def test_state_minimal(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for State with minimal values.""" + zero_cp = _zero_checkpoint() + zero_header = BlockHeader( + slot=Slot(0), + proposer_index=ValidatorIndex(0), + parent_root=Bytes32.zero(), + state_root=Bytes32.zero(), + body_root=Bytes32.zero(), + ) + ssz( + type_name="State", + value=State( + config=Config(genesis_time=Uint64(0)), + slot=Slot(0), + latest_block_header=zero_header, + latest_justified=zero_cp, + latest_finalized=zero_cp, + historical_block_hashes=HistoricalBlockHashes(data=[]), + justified_slots=JustifiedSlots(data=[]), + validators=Validators(data=[Validator(pubkey=Bytes52.zero(), index=ValidatorIndex(0))]), + justifications_roots=JustificationRoots(data=[]), + justifications_validators=JustificationValidators(data=[]), + ), + ) + + +def test_state_with_validators(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for State with multiple validators and history.""" + ssz( + type_name="State", + value=State( + config=Config(genesis_time=Uint64(1609459200)), + slot=Slot(100), + latest_block_header=BlockHeader( + slot=Slot(99), + proposer_index=ValidatorIndex(2), + parent_root=Bytes32(b"\x01" * 32), + state_root=Bytes32(b"\x02" * 32), + body_root=Bytes32(b"\x03" * 32), + ), + latest_justified=Checkpoint(root=Bytes32(b"\x04" * 32), slot=Slot(64)), + latest_finalized=Checkpoint(root=Bytes32(b"\x05" * 32), slot=Slot(32)), + historical_block_hashes=HistoricalBlockHashes( + data=[Bytes32(b"\x10" * 32), Bytes32(b"\x11" * 32)] + ), + justified_slots=JustifiedSlots(data=[Boolean(True), Boolean(False), Boolean(True)]), + validators=Validators( + data=[ + Validator(pubkey=Bytes52(b"\x01" * 52), index=ValidatorIndex(0)), + Validator(pubkey=Bytes52(b"\x02" * 52), index=ValidatorIndex(1)), + Validator(pubkey=Bytes52(b"\x03" * 52), index=ValidatorIndex(2)), + Validator(pubkey=Bytes52(b"\x04" * 52), index=ValidatorIndex(3)), + ] + ), + justifications_roots=JustificationRoots(data=[Bytes32(b"\x20" * 32)]), + justifications_validators=JustificationValidators( + data=[Boolean(True), Boolean(True), Boolean(False), Boolean(True)] + ), + ), + ) diff --git a/tests/consensus/devnet/ssz/test_networking_containers.py b/tests/consensus/devnet/ssz/test_networking_containers.py new file mode 100644 index 00000000..21435f13 --- /dev/null +++ b/tests/consensus/devnet/ssz/test_networking_containers.py @@ -0,0 +1,70 @@ +"""SSZ conformance tests for networking containers.""" + +import pytest +from consensus_testing import SSZTestFiller + +from lean_spec.subspecs.containers import Checkpoint, Slot +from lean_spec.subspecs.networking.reqresp.message import ( + BlocksByRootRequest, + RequestedBlockRoots, + Status, +) +from lean_spec.types import Bytes32 + +pytestmark = pytest.mark.valid_until("Devnet") + + +# --- Status --- + + +def test_status_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Status with zero values.""" + ssz( + type_name="Status", + value=Status( + finalized=Checkpoint(root=Bytes32.zero(), slot=Slot(0)), + head=Checkpoint(root=Bytes32.zero(), slot=Slot(0)), + ), + ) + + +def test_status_typical(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Status with typical values.""" + ssz( + type_name="Status", + value=Status( + finalized=Checkpoint(root=Bytes32(b"\x01" * 32), slot=Slot(100)), + head=Checkpoint(root=Bytes32(b"\x02" * 32), slot=Slot(150)), + ), + ) + + +# --- BlocksByRootRequest --- + + +def test_blocks_by_root_request_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlocksByRootRequest with no roots.""" + ssz( + type_name="BlocksByRootRequest", + value=BlocksByRootRequest(roots=RequestedBlockRoots(data=[])), + ) + + +def test_blocks_by_root_request_single(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlocksByRootRequest with single root.""" + ssz( + type_name="BlocksByRootRequest", + value=BlocksByRootRequest(roots=RequestedBlockRoots(data=[Bytes32(b"\xab" * 32)])), + ) + + +def test_blocks_by_root_request_multiple(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for BlocksByRootRequest with multiple roots.""" + ssz( + type_name="BlocksByRootRequest", + value=BlocksByRootRequest( + roots=RequestedBlockRoots( + data=[Bytes32(b"\x01" * 32), Bytes32(b"\x02" * 32), Bytes32(b"\x03" * 32)] + ) + ), + ) diff --git a/tests/consensus/devnet/ssz/test_xmss_containers.py b/tests/consensus/devnet/ssz/test_xmss_containers.py new file mode 100644 index 00000000..f29c7945 --- /dev/null +++ b/tests/consensus/devnet/ssz/test_xmss_containers.py @@ -0,0 +1,230 @@ +"""SSZ conformance tests for XMSS containers.""" + +import pytest +from consensus_testing import SSZTestFiller + +from lean_spec.subspecs.containers.attestation import AggregationBits +from lean_spec.subspecs.koalabear import Fp +from lean_spec.subspecs.xmss import PublicKey, SecretKey, Signature +from lean_spec.subspecs.xmss.aggregation import AggregatedSignatureProof +from lean_spec.subspecs.xmss.subtree import HashSubTree +from lean_spec.subspecs.xmss.types import ( + HASH_DIGEST_LENGTH, + HashDigestList, + HashDigestVector, + HashTreeLayer, + HashTreeLayers, + HashTreeOpening, + Parameter, + PRFKey, + Randomness, +) +from lean_spec.types import Boolean, Uint64 +from lean_spec.types.byte_arrays import ByteListMiB + +pytestmark = pytest.mark.valid_until("Devnet") + + +# --- Helper functions --- + + +def _zero_hash_digest_vector() -> HashDigestVector: + return HashDigestVector(data=[Fp(0) for _ in range(HASH_DIGEST_LENGTH)]) + + +def _zero_randomness() -> Randomness: + return Randomness(data=[Fp(0) for _ in range(Randomness.LENGTH)]) + + +def _zero_parameter() -> Parameter: + return Parameter(data=[Fp(0) for _ in range(Parameter.LENGTH)]) + + +# --- PublicKey --- + + +def test_public_key_zero(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for PublicKey with zero values.""" + ssz( + type_name="PublicKey", + value=PublicKey(root=_zero_hash_digest_vector(), parameter=_zero_parameter()), + ) + + +# --- Signature --- + + +def test_signature_empty_path(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Signature with empty authentication path.""" + ssz( + type_name="Signature", + value=Signature( + path=HashTreeOpening(siblings=HashDigestList(data=[])), + rho=_zero_randomness(), + hashes=HashDigestList(data=[]), + ), + ) + + +def test_signature_with_siblings(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for Signature with authentication path siblings.""" + ssz( + type_name="Signature", + value=Signature( + path=HashTreeOpening( + siblings=HashDigestList( + data=[_zero_hash_digest_vector(), _zero_hash_digest_vector()] + ) + ), + rho=_zero_randomness(), + hashes=HashDigestList(data=[_zero_hash_digest_vector()]), + ), + ) + + +# --- SecretKey --- + + +def test_secret_key_minimal(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for SecretKey with minimal values.""" + empty_subtree = HashSubTree( + depth=Uint64(4), + lowest_layer=Uint64(0), + layers=HashTreeLayers(data=[]), + ) + ssz( + type_name="SecretKey", + value=SecretKey( + prf_key=PRFKey.zero(), + parameter=_zero_parameter(), + activation_epoch=Uint64(0), + num_active_epochs=Uint64(1), + top_tree=empty_subtree, + left_bottom_tree_index=Uint64(0), + left_bottom_tree=empty_subtree, + right_bottom_tree=empty_subtree, + ), + ) + + +# --- HashTreeOpening --- + + +def test_hash_tree_opening_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashTreeOpening with no siblings.""" + ssz( + type_name="HashTreeOpening", + value=HashTreeOpening(siblings=HashDigestList(data=[])), + ) + + +def test_hash_tree_opening_single(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashTreeOpening with single sibling.""" + ssz( + type_name="HashTreeOpening", + value=HashTreeOpening(siblings=HashDigestList(data=[_zero_hash_digest_vector()])), + ) + + +def test_hash_tree_opening_multiple(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashTreeOpening with multiple siblings.""" + ssz( + type_name="HashTreeOpening", + value=HashTreeOpening( + siblings=HashDigestList( + data=[ + _zero_hash_digest_vector(), + _zero_hash_digest_vector(), + _zero_hash_digest_vector(), + ] + ) + ), + ) + + +# --- HashTreeLayer --- + + +def test_hash_tree_layer_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashTreeLayer with no nodes.""" + ssz( + type_name="HashTreeLayer", + value=HashTreeLayer(start_index=Uint64(0), nodes=HashDigestList(data=[])), + ) + + +def test_hash_tree_layer_single(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashTreeLayer with single node.""" + ssz( + type_name="HashTreeLayer", + value=HashTreeLayer( + start_index=Uint64(0), + nodes=HashDigestList(data=[_zero_hash_digest_vector()]), + ), + ) + + +def test_hash_tree_layer_multiple(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashTreeLayer with multiple nodes.""" + ssz( + type_name="HashTreeLayer", + value=HashTreeLayer( + start_index=Uint64(4), + nodes=HashDigestList(data=[_zero_hash_digest_vector(), _zero_hash_digest_vector()]), + ), + ) + + +# --- HashSubTree --- + + +def test_hash_subtree_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashSubTree with no layers.""" + ssz( + type_name="HashSubTree", + value=HashSubTree(depth=Uint64(4), lowest_layer=Uint64(0), layers=HashTreeLayers(data=[])), + ) + + +def test_hash_subtree_with_layers(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for HashSubTree with layers.""" + ssz( + type_name="HashSubTree", + value=HashSubTree( + depth=Uint64(4), + lowest_layer=Uint64(0), + layers=HashTreeLayers( + data=[ + HashTreeLayer( + start_index=Uint64(0), + nodes=HashDigestList(data=[_zero_hash_digest_vector()]), + ) + ] + ), + ), + ) + + +# --- AggregatedSignatureProof --- + + +def test_aggregated_signature_proof_empty(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for AggregatedSignatureProof with empty proof data.""" + ssz( + type_name="AggregatedSignatureProof", + value=AggregatedSignatureProof( + participants=AggregationBits(data=[Boolean(True)]), + proof_data=ByteListMiB(data=b""), + ), + ) + + +def test_aggregated_signature_proof_with_data(ssz: SSZTestFiller) -> None: + """SSZ roundtrip for AggregatedSignatureProof with proof data.""" + ssz( + type_name="AggregatedSignatureProof", + value=AggregatedSignatureProof( + participants=AggregationBits(data=[Boolean(True), Boolean(False), Boolean(True)]), + proof_data=ByteListMiB(data=b"\xde\xad\xbe\xef"), + ), + ) From 87a2b7892b8155fcdfe25936906d20e5f49c623b Mon Sep 17 00:00:00 2001 From: unnawut Date: Tue, 10 Feb 2026 16:31:52 +0700 Subject: [PATCH 2/3] test vectors should serialize Signature to raw bytes --- src/lean_spec/subspecs/xmss/containers.py | 7 ++++ .../devnet/ssz/test_consensus_containers.py | 28 ++++++------- .../devnet/ssz/test_xmss_containers.py | 40 +++++++------------ 3 files changed, 34 insertions(+), 41 deletions(-) diff --git a/src/lean_spec/subspecs/xmss/containers.py b/src/lean_spec/subspecs/xmss/containers.py index e7932cb2..d2d0ea6a 100644 --- a/src/lean_spec/subspecs/xmss/containers.py +++ b/src/lean_spec/subspecs/xmss/containers.py @@ -9,6 +9,8 @@ from typing import TYPE_CHECKING, Mapping, NamedTuple +from pydantic import model_serializer + from ...types import Bytes32, Uint64 from ...types.container import Container from .subtree import HashSubTree @@ -67,6 +69,11 @@ class Signature(Container): hashes: HashDigestList """The one-time signature itself: a list of intermediate Winternitz chain hashes.""" + @model_serializer(mode="plain", when_used="json") + def _serialize_as_bytes(self) -> str: + """Serialize as hex-encoded SSZ bytes for JSON output.""" + return "0x" + self.encode_bytes().hex() + def verify( self, public_key: PublicKey, diff --git a/tests/consensus/devnet/ssz/test_consensus_containers.py b/tests/consensus/devnet/ssz/test_consensus_containers.py index 40f301ad..9a1ba7aa 100644 --- a/tests/consensus/devnet/ssz/test_consensus_containers.py +++ b/tests/consensus/devnet/ssz/test_consensus_containers.py @@ -33,14 +33,8 @@ JustifiedSlots, Validators, ) -from lean_spec.subspecs.koalabear import Fp from lean_spec.subspecs.xmss import Signature from lean_spec.subspecs.xmss.aggregation import AggregatedSignatureProof -from lean_spec.subspecs.xmss.types import ( - HashDigestList, - HashTreeOpening, - Randomness, -) from lean_spec.types import Boolean, Bytes32, Bytes52, Uint64 from lean_spec.types.byte_arrays import ByteListMiB @@ -66,12 +60,14 @@ def _typical_attestation_data() -> AttestationData: return AttestationData(slot=Slot(100), head=head, target=target, source=source) -def _zero_signature() -> Signature: - return Signature( - path=HashTreeOpening(siblings=HashDigestList(data=[])), - rho=Randomness(data=[Fp(0) for _ in range(Randomness.LENGTH)]), - hashes=HashDigestList(data=[]), - ) +# Empty signature: path=[], rho=zeros, hashes=[] +EMPTY_SIGNATURE_BYTES = bytes.fromhex( + "24000000000000000000000000000000000000000000000000000000000000002800000004000000" +) + + +def _empty_signature() -> Signature: + return Signature.decode_bytes(EMPTY_SIGNATURE_BYTES) # --- Checkpoint --- @@ -132,7 +128,7 @@ def test_signed_attestation_minimal(ssz: SSZTestFiller) -> None: value=SignedAttestation( validator_id=ValidatorIndex(0), message=_zero_attestation_data(), - signature=_zero_signature(), + signature=_empty_signature(), ), ) @@ -291,7 +287,7 @@ def test_block_signatures_empty(ssz: SSZTestFiller) -> None: type_name="BlockSignatures", value=BlockSignatures( attestation_signatures=AttestationSignatures(data=[]), - proposer_signature=_zero_signature(), + proposer_signature=_empty_signature(), ), ) @@ -309,7 +305,7 @@ def test_block_signatures_with_attestation(ssz: SSZTestFiller) -> None: ) ] ), - proposer_signature=_zero_signature(), + proposer_signature=_empty_signature(), ), ) @@ -330,7 +326,7 @@ def test_signed_block_with_attestation_minimal(ssz: SSZTestFiller) -> None: message = BlockWithAttestation(block=block, proposer_attestation=attestation) signature = BlockSignatures( attestation_signatures=AttestationSignatures(data=[]), - proposer_signature=_zero_signature(), + proposer_signature=_empty_signature(), ) ssz( type_name="SignedBlockWithAttestation", diff --git a/tests/consensus/devnet/ssz/test_xmss_containers.py b/tests/consensus/devnet/ssz/test_xmss_containers.py index f29c7945..16b0cd02 100644 --- a/tests/consensus/devnet/ssz/test_xmss_containers.py +++ b/tests/consensus/devnet/ssz/test_xmss_containers.py @@ -17,7 +17,6 @@ HashTreeOpening, Parameter, PRFKey, - Randomness, ) from lean_spec.types import Boolean, Uint64 from lean_spec.types.byte_arrays import ByteListMiB @@ -32,10 +31,6 @@ def _zero_hash_digest_vector() -> HashDigestVector: return HashDigestVector(data=[Fp(0) for _ in range(HASH_DIGEST_LENGTH)]) -def _zero_randomness() -> Randomness: - return Randomness(data=[Fp(0) for _ in range(Randomness.LENGTH)]) - - def _zero_parameter() -> Parameter: return Parameter(data=[Fp(0) for _ in range(Parameter.LENGTH)]) @@ -53,33 +48,28 @@ def test_public_key_zero(ssz: SSZTestFiller) -> None: # --- Signature --- +# Empty path: path=[], rho=zeros, hashes=[] +SIGNATURE_EMPTY_PATH = bytes.fromhex( + "24000000000000000000000000000000000000000000000000000000000000002800000004000000" +) + +# With siblings: path=[zero, zero], rho=zeros, hashes=[zero] +SIGNATURE_WITH_SIBLINGS = bytes.fromhex( + "24000000000000000000000000000000000000000000000000000000000000006800000004000000" + "0000000000000000000000000000000000000000000000000000000000000000" + "0000000000000000000000000000000000000000000000000000000000000000" + "0000000000000000000000000000000000000000000000000000000000000000" +) + def test_signature_empty_path(ssz: SSZTestFiller) -> None: """SSZ roundtrip for Signature with empty authentication path.""" - ssz( - type_name="Signature", - value=Signature( - path=HashTreeOpening(siblings=HashDigestList(data=[])), - rho=_zero_randomness(), - hashes=HashDigestList(data=[]), - ), - ) + ssz(type_name="Signature", value=Signature.decode_bytes(SIGNATURE_EMPTY_PATH)) def test_signature_with_siblings(ssz: SSZTestFiller) -> None: """SSZ roundtrip for Signature with authentication path siblings.""" - ssz( - type_name="Signature", - value=Signature( - path=HashTreeOpening( - siblings=HashDigestList( - data=[_zero_hash_digest_vector(), _zero_hash_digest_vector()] - ) - ), - rho=_zero_randomness(), - hashes=HashDigestList(data=[_zero_hash_digest_vector()]), - ), - ) + ssz(type_name="Signature", value=Signature.decode_bytes(SIGNATURE_WITH_SIBLINGS)) # --- SecretKey --- From acac500130f91e0cc17f3d1d9a046796ec868489 Mon Sep 17 00:00:00 2001 From: unnawut Date: Wed, 11 Feb 2026 17:25:18 +0700 Subject: [PATCH 3/3] remove hash tree root checks (not used with all containers) --- .../consensus_testing/test_fixtures/ssz.py | 29 +++++-------------- 1 file changed, 7 insertions(+), 22 deletions(-) diff --git a/packages/testing/src/consensus_testing/test_fixtures/ssz.py b/packages/testing/src/consensus_testing/test_fixtures/ssz.py index 08cfb763..9a2fd176 100644 --- a/packages/testing/src/consensus_testing/test_fixtures/ssz.py +++ b/packages/testing/src/consensus_testing/test_fixtures/ssz.py @@ -4,8 +4,6 @@ from pydantic import field_serializer -from lean_spec.subspecs.ssz.hash import hash_tree_root -from lean_spec.types import Bytes32 from lean_spec.types.container import Container from .base import BaseConsensusFixture @@ -15,17 +13,16 @@ class SSZTest(BaseConsensusFixture): """ Test fixture for SSZ serialization/deserialization conformance. - Tests roundtrip serialization and hash tree root computation for SSZ containers. + Tests roundtrip serialization for SSZ containers. Structure: type_name: Name of the container class value: The container instance serialized: Hex-encoded SSZ bytes (computed) - root: Hex-encoded hash tree root (computed) """ format_name: ClassVar[str] = "ssz" - description: ClassVar[str] = "Tests SSZ serialization roundtrip and hash tree root computation" + description: ClassVar[str] = "Tests SSZ serialization roundtrip" type_name: str """Name of the container class being tested.""" @@ -36,9 +33,6 @@ class SSZTest(BaseConsensusFixture): serialized: str = "" """Hex-encoded SSZ serialized bytes (computed during make_fixture).""" - root: str = "" - """Hex-encoded hash tree root (computed during make_fixture).""" - @field_serializer("value", when_used="json") def serialize_value(self, value: Container) -> dict[str, Any]: """Serialize the container value to JSON using its native serialization.""" @@ -46,18 +40,17 @@ def serialize_value(self, value: Container) -> dict[str, Any]: def make_fixture(self) -> "SSZTest": """ - Generate the fixture by testing SSZ roundtrip and computing root. + Generate the fixture by testing SSZ roundtrip. 1. Serialize the value to SSZ bytes 2. Deserialize the bytes back to a container 3. Verify the roundtrip produces the same value - 4. Compute the hash tree root Returns: - SSZTest with computed serialized and root fields. + SSZTest with computed serialized field. Raises: - AssertionError: If roundtrip fails or types don't match. + AssertionError: If roundtrip fails. """ # Serialize to SSZ bytes ssz_bytes = self.value.encode_bytes() @@ -74,13 +67,5 @@ def make_fixture(self) -> "SSZTest": f"Decoded: {decoded}" ) - # Compute hash tree root - htr: Bytes32 = hash_tree_root(self.value) - - # Return fixture with computed fields - return self.model_copy( - update={ - "serialized": "0x" + ssz_bytes.hex(), - "root": "0x" + htr.hex(), - } - ) + # Return fixture with computed serialized field + return self.model_copy(update={"serialized": "0x" + ssz_bytes.hex()})