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..9a2fd176 --- /dev/null +++ b/packages/testing/src/consensus_testing/test_fixtures/ssz.py @@ -0,0 +1,71 @@ +"""SSZ test fixture format for serialization conformance testing.""" + +from typing import Any, ClassVar + +from pydantic import field_serializer + +from lean_spec.types.container import Container + +from .base import BaseConsensusFixture + + +class SSZTest(BaseConsensusFixture): + """ + Test fixture for SSZ serialization/deserialization conformance. + + Tests roundtrip serialization for SSZ containers. + + Structure: + type_name: Name of the container class + value: The container instance + serialized: Hex-encoded SSZ bytes (computed) + """ + + format_name: ClassVar[str] = "ssz" + description: ClassVar[str] = "Tests SSZ serialization roundtrip" + + 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).""" + + @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. + + 1. Serialize the value to SSZ bytes + 2. Deserialize the bytes back to a container + 3. Verify the roundtrip produces the same value + + Returns: + SSZTest with computed serialized field. + + Raises: + AssertionError: If roundtrip fails. + """ + # 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}" + ) + + # Return fixture with computed serialized field + return self.model_copy(update={"serialized": "0x" + ssz_bytes.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/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/__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..9a1ba7aa --- /dev/null +++ b/tests/consensus/devnet/ssz/test_consensus_containers.py @@ -0,0 +1,432 @@ +"""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.xmss import Signature +from lean_spec.subspecs.xmss.aggregation import AggregatedSignatureProof +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) + + +# Empty signature: path=[], rho=zeros, hashes=[] +EMPTY_SIGNATURE_BYTES = bytes.fromhex( + "24000000000000000000000000000000000000000000000000000000000000002800000004000000" +) + + +def _empty_signature() -> Signature: + return Signature.decode_bytes(EMPTY_SIGNATURE_BYTES) + + +# --- 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=_empty_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=_empty_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=_empty_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=_empty_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..16b0cd02 --- /dev/null +++ b/tests/consensus/devnet/ssz/test_xmss_containers.py @@ -0,0 +1,220 @@ +"""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, +) +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_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 --- + +# 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.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.decode_bytes(SIGNATURE_WITH_SIBLINGS)) + + +# --- 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"), + ), + )