refactor(lstar): freeze the Block family (leanEthereum#843)

* refactor(lstar): freeze constructed-once SSZ containers

Set frozen=True on the lstar containers that are built once and never
mutated after construction: GenesisConfig, Validator, AggregatedAttestation,
and SignedAggregatedAttestation. This completes the freeze begun on
Checkpoint, AttestationData, Attestation, SingleMessageAggregate, and
MultiMessageAggregate, so accidental post-construction mutation now raises
instead of silently succeeding.

Freezing Validator required one call-site change: the fork-choice test
fixture injected public keys by mutating each validator in place. It now
rebuilds each validator with model_copy(update=...), which is byte-identical
in hash_tree_root and leaves the originals untouched.

Block, BlockHeader, BlockBody, and SignedBlock stay mutable for now. Their
state_root and body are patched in place during block production and the
state transition (block_production.py and state_transition.py), a two-pass
pattern entangled with the larger state-transition refactor. State and Store
remain mutable by design.

Frozen is an enforcement aid, not a hard immutability guarantee: model_config
is itself a mutable dict and an unfreeze-then-assign sequence can still leave
a field mutable (pydantic#12361). It catches accidental mutation; it is not a
soundness boundary.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor(lstar): freeze the Block family

Set frozen=True on BlockBody, BlockHeader, Block, and SignedBlock, the last
mutable containers in the lstar object graph apart from State and Store. With
this, every consensus container except the State accumulator and the
fork-choice Store is immutable: accidental post-construction mutation now
raises instead of silently corrupting a value that may already be referenced
elsewhere or committed to a root.

These four were deferred from the initial freeze because each was patched in
place after construction. Every such site is migrated to model_copy(update=...),
which is byte-identical in hash_tree_root:

  - Block production builds the block with a zero state root, processes it,
    then rebuilds the block with the real root.
  - The slot transition rebuilds the latest header with its cached state root
    and reassigns it on the still-mutable State, instead of mutating the header
    in place.
  - The test-framework builders and the signature-tampering helpers rebuild
    blocks and bodies through layered model_copy, matching the pattern the
    attestation-swap tamper already used.

State and Store stay mutable. State is the deepcopy-and-mutate accumulator at
the heart of the state transition; converting it to return a new value is a
larger, separate change. Store is fork-choice scratch state.

As with the earlier container freeze, frozen is an enforcement aid, not a hard
immutability guarantee (model_config is a mutable dict; pydantic#12361).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: alexanderlhicks

packages/testing/src/consensus_testing/test_fixtures/fork_choice.py

@@ -241,7 +241,9 @@ def make_fixture(self) -> Self:
         # Updating validators changes the state root.
         # We must also update the anchor block to match.
         self.anchor_state.validators = Validators(data=updated_validators)
-        self.anchor_block.state_root = hash_tree_root(self.anchor_state)
+        self.anchor_block = self.anchor_block.model_copy(
+            update={"state_root": hash_tree_root(self.anchor_state)}
+        )
 
         # Store initialization
         #

packages/testing/src/consensus_testing/test_fixtures/state_transition.py

@@ -278,16 +278,24 @@ def _build_block_from_spec(
                 )
                 for fa in spec.forced_attestations
             ]
-            block.body.attestations = AggregatedAttestations(
-                data=[*block.body.attestations.data, *forced]
+            block = block.model_copy(
+                update={
+                    "body": block.body.model_copy(
+                        update={
+                            "attestations": AggregatedAttestations(
+                                data=[*block.body.attestations.data, *forced]
+                            )
+                        }
+                    )
+                }
             )
 
             # The body changed, so re-run the transition to get the correct
             # post-state and state root.
             if post_state is not None:
                 post_state = LstarSpec().process_slots(state, spec.slot)
                 post_state = LstarSpec().process_block(post_state, block)
-                block.state_root = hash_tree_root(post_state)
+                block = block.model_copy(update={"state_root": hash_tree_root(post_state)})
 
         return block, post_state
 

packages/testing/src/consensus_testing/test_fixtures/verify_signatures.py

@@ -165,8 +165,13 @@ def _apply_tamper(self, signed_block: SignedBlock) -> SignedBlock:
             value = self.tamper.get("value")
             if value is None:
                 raise ValueError("set_proposer_index requires a value")
-            signed_block.block.proposer_index = ValidatorIndex(int(value))
-            return signed_block
+            return signed_block.model_copy(
+                update={
+                    "block": signed_block.block.model_copy(
+                        update={"proposer_index": ValidatorIndex(int(value))}
+                    )
+                }
+            )
 
         if operation == "clear_first_attestation_bits":
             original = signed_block.block.body.attestations.data
@@ -175,18 +180,32 @@ def _apply_tamper(self, signed_block: SignedBlock) -> SignedBlock:
             first = original[0]
             empty_bits = AggregationBits(data=[Boolean(False)] * len(first.aggregation_bits.data))
             cleared = AggregatedAttestation(aggregation_bits=empty_bits, data=first.data)
-            signed_block.block.body.attestations = AggregatedAttestations(
-                data=[cleared, *original[1:]]
+            return signed_block.model_copy(
+                update={
+                    "block": signed_block.block.model_copy(
+                        update={
+                            "body": signed_block.block.body.model_copy(
+                                update={
+                                    "attestations": AggregatedAttestations(
+                                        data=[cleared, *original[1:]]
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
             )
-            return signed_block
 
         if operation == "corrupt_proof":
             # Replace the merged proof with a short bogus payload.
             # The verifier rejects the malformed proof bytes.
-            signed_block.proof = MultiMessageAggregate(
-                proof=ByteList512KiB(data=b"\x00\x01\x02\x03"),
+            return signed_block.model_copy(
+                update={
+                    "proof": MultiMessageAggregate(
+                        proof=ByteList512KiB(data=b"\x00\x01\x02\x03"),
+                    )
+                }
             )
-            return signed_block
 
         if operation == "append_phantom_attestation":
             # Add a body attestation with no matching proof component.
@@ -203,19 +222,35 @@ def _apply_tamper(self, signed_block: SignedBlock) -> SignedBlock:
                 aggregation_bits=AggregationBits(data=[Boolean(True)]),
                 data=phantom_data,
             )
-            signed_block.block.body.attestations = AggregatedAttestations(
-                data=[*signed_block.block.body.attestations.data, phantom]
+            return signed_block.model_copy(
+                update={
+                    "block": signed_block.block.model_copy(
+                        update={
+                            "body": signed_block.block.body.model_copy(
+                                update={
+                                    "attestations": AggregatedAttestations(
+                                        data=[*signed_block.block.body.attestations.data, phantom]
+                                    )
+                                }
+                            )
+                        }
+                    )
+                }
             )
-            return signed_block
 
         if operation == "mutate_state_root":
             # Change a block field after signing so the block root differs.
             # The proposer component's bound message no longer matches the
             # recomputed block root, even though the signature is honest.
             # This is the repackaging vector: an honest proof reused under
             # a different message.
-            signed_block.block.state_root = Bytes32(b"\xff" * 32)
-            return signed_block
+            return signed_block.model_copy(
+                update={
+                    "block": signed_block.block.model_copy(
+                        update={"state_root": Bytes32(b"\xff" * 32)}
+                    )
+                }
+            )
 
         if operation == "swap_first_two_attestations":
             body = signed_block.block.body

packages/testing/src/consensus_testing/test_types/aggregated_attestation_spec.py

@@ -202,8 +202,16 @@ def build_invalid_proof(
             invalid_proof = SingleMessageAggregate(participants=aggregation_bits, proof=placeholder)
 
         # Append invalid attestation to the block body.
-        block.body.attestations = AggregatedAttestations(
-            data=[*block.body.attestations.data, invalid_aggregated]
+        block = block.model_copy(
+            update={
+                "body": block.body.model_copy(
+                    update={
+                        "attestations": AggregatedAttestations(
+                            data=[*block.body.attestations.data, invalid_aggregated]
+                        )
+                    }
+                )
+            }
         )
 
         return block, invalid_proof

packages/testing/src/consensus_testing/test_types/block_spec.py

@@ -548,22 +548,30 @@ def build_signed_block_with_store(
                     attestation_spec.validator_indices, attestation_data
                 )
                 block_proofs.append(proof)
-                final_block.body.attestations = AggregatedAttestations(
-                    data=[
-                        *final_block.body.attestations.data,
-                        AggregatedAttestation(
-                            aggregation_bits=AggregationBits.from_indices(
-                                attestation_spec.validator_indices
-                            ),
-                            data=attestation_data,
-                        ),
-                    ]
+                final_block = final_block.model_copy(
+                    update={
+                        "body": final_block.body.model_copy(
+                            update={
+                                "attestations": AggregatedAttestations(
+                                    data=[
+                                        *final_block.body.attestations.data,
+                                        AggregatedAttestation(
+                                            aggregation_bits=AggregationBits.from_indices(
+                                                attestation_spec.validator_indices
+                                            ),
+                                            data=attestation_data,
+                                        ),
+                                    ]
+                                )
+                            }
+                        )
+                    }
                 )
 
             # Recompute state root with the modified body.
             post_state = spec.process_slots(parent_state, self.slot)
             post_state = spec.process_block(post_state, final_block)
-            final_block.state_root = hash_tree_root(post_state)
+            final_block = final_block.model_copy(update={"state_root": hash_tree_root(post_state)})
 
         signed_block = self._sign_block(
             final_block, block_proofs, proposer_index, key_manager, parent_state

src/lean_spec/spec/forks/lstar/block_production.py

@@ -282,6 +282,6 @@ def build_block(
         # Merging proofs keeps the same voters, so the post-state is unchanged.
         # Only the body's shape differs, so just the root is needed.
         post_state = self.process_block(self.process_slots(state, slot), final_block)
-        final_block.state_root = hash_tree_root(post_state)
+        final_block = final_block.model_copy(update={"state_root": hash_tree_root(post_state)})
 
         return final_block, post_state, aggregated_attestations, aggregated_signatures

src/lean_spec/spec/forks/lstar/containers/block.py

@@ -10,6 +10,8 @@
 class BlockBody(Container):
     """Payload of a block containing attestations."""
 
+    model_config = Container.model_config | {"frozen": True}
+
     attestations: AggregatedAttestations
     """Attestations in the block. Signatures are folded into the block-level proof."""
 
@@ -22,6 +24,8 @@ class BlockHeader(Container):
     Smaller than full blocks.
     """
 
+    model_config = Container.model_config | {"frozen": True}
+
     slot: Slot
     """The slot in which the block was proposed."""
 
@@ -41,6 +45,8 @@ class BlockHeader(Container):
 class Block(Container):
     """A complete block including header and body."""
 
+    model_config = Container.model_config | {"frozen": True}
+
     slot: Slot
     """The slot in which the block was proposed."""
 
@@ -65,6 +71,8 @@ class SignedBlock(Container):
     over the block root.
     """
 
+    model_config = Container.model_config | {"frozen": True}
+
     block: Block
     """The block being signed."""
 

src/lean_spec/spec/forks/lstar/state_transition.py

@@ -144,7 +144,9 @@ def process_slots(self, state: State, target_slot: Slot) -> State:
             )
 
             if needs_state_root:
-                state.latest_block_header.state_root = cached_state_root
+                state.latest_block_header = state.latest_block_header.model_copy(
+                    update={"state_root": cached_state_root}
+                )
             state.slot = Slot(state.slot + Slot(1))
 
         # Reached the target slot. Return the advanced state.

tests/consensus/lstar/fc/test_checkpoint_sync.py

@@ -361,8 +361,7 @@ def test_store_from_anchor_rejects_mismatched_state_root(
     # mismatch below the only difference between this test and the happy path.
     assert anchor_block.state_root == hash_tree_root(anchor_state)
 
-    anchor_block.state_root = Bytes32(b"\xff" * 32)
-    bad_anchor_block = anchor_block
+    bad_anchor_block = anchor_block.model_copy(update={"state_root": Bytes32(b"\xff" * 32)})
 
     fork_choice_test(
         anchor_state=anchor_state,

tests/lean_spec/spec/forks/lstar/containers/test_block.py

@@ -0,0 +1,84 @@
+"""Tests for the Block container family."""
+
+import pytest
+from pydantic import ValidationError
+
+from lean_spec.spec.forks import Slot, ValidatorIndex
+from lean_spec.spec.forks.lstar.containers import (
+    AggregatedAttestations,
+    Block,
+    BlockBody,
+    BlockHeader,
+    MultiMessageAggregate,
+    SignedBlock,
+)
+from lean_spec.spec.ssz import ByteList512KiB, Bytes32
+
+
+def _empty_body() -> BlockBody:
+    """A block body carrying no attestations."""
+    return BlockBody(attestations=AggregatedAttestations(data=[]))
+
+
+def _block() -> Block:
+    """A minimal block with a zeroed state root and an empty body."""
+    return Block(
+        slot=Slot(1),
+        proposer_index=ValidatorIndex(0),
+        parent_root=Bytes32.zero(),
+        state_root=Bytes32.zero(),
+        body=_empty_body(),
+    )
+
+
+class TestBlockBodyImmutability:
+    """Frozen-model semantics forbid post-construction mutation."""
+
+    def test_assigning_attestations_raises(self) -> None:
+        """Assigning new attestations on a constructed body raises."""
+        body = _empty_body()
+        with pytest.raises(ValidationError, match="frozen"):
+            body.attestations = AggregatedAttestations(data=[])
+
+
+class TestBlockHeaderImmutability:
+    """Frozen-model semantics forbid post-construction mutation."""
+
+    def test_assigning_state_root_raises(self) -> None:
+        """Assigning a new state root on a constructed header raises."""
+        header = BlockHeader(
+            slot=Slot(1),
+            proposer_index=ValidatorIndex(0),
+            parent_root=Bytes32.zero(),
+            state_root=Bytes32.zero(),
+            body_root=Bytes32.zero(),
+        )
+        with pytest.raises(ValidationError, match="frozen"):
+            header.state_root = Bytes32(b"\xff" * 32)
+
+
+class TestBlockImmutability:
+    """Frozen-model semantics forbid post-construction mutation."""
+
+    def test_assigning_state_root_raises(self) -> None:
+        """Assigning a new state root on a constructed block raises."""
+        block = _block()
+        with pytest.raises(ValidationError, match="frozen"):
+            block.state_root = Bytes32(b"\xff" * 32)
+
+    def test_assigning_body_raises(self) -> None:
+        """Assigning a new body on a constructed block raises."""
+        block = _block()
+        with pytest.raises(ValidationError, match="frozen"):
+            block.body = _empty_body()
+
+
+class TestSignedBlockImmutability:
+    """Frozen-model semantics forbid post-construction mutation."""
+
+    def test_assigning_proof_raises(self) -> None:
+        """Assigning a new proof on a constructed signed block raises."""
+        proof = MultiMessageAggregate(proof=ByteList512KiB(data=b""))
+        signed_block = SignedBlock(block=_block(), proof=proof)
+        with pytest.raises(ValidationError, match="frozen"):
+            signed_block.proof = proof