diff --git a/src/lean_spec/node/storage/database.py b/src/lean_spec/node/storage/database.py index 6b8ab5f5e..3023a3aed 100644 --- a/src/lean_spec/node/storage/database.py +++ b/src/lean_spec/node/storage/database.py @@ -6,7 +6,7 @@ from contextlib import contextmanager from typing import Protocol -from lean_spec.spec.forks import Checkpoint, Slot +from lean_spec.spec.forks import Checkpoint, Slot, ValidatorIndex from lean_spec.spec.forks.protocol import ( SpecBlockType, SpecStateType, @@ -111,6 +111,26 @@ def put_genesis_time(self, genesis_time: Uint64) -> None: """Store the genesis time as a Unix timestamp.""" ... + # Signing Records + + def get_last_signed_slot(self, validator_index: ValidatorIndex, key_role: str) -> Slot | None: + """ + Retrieve the highest slot one of a validator's keys has signed. + + Returns None when that key has not signed yet. + The key role separates a validator's attestation key from its proposal key. + """ + ... + + def put_last_signed_slot( + self, + validator_index: ValidatorIndex, + key_role: str, + slot: Slot, + ) -> None: + """Record the highest slot one of a validator's keys has signed.""" + ... + # Transaction Control @contextmanager diff --git a/src/lean_spec/node/storage/namespaces.py b/src/lean_spec/node/storage/namespaces.py index 90ea84e4e..f0e9c2316 100644 --- a/src/lean_spec/node/storage/namespaces.py +++ b/src/lean_spec/node/storage/namespaces.py @@ -77,3 +77,20 @@ block_root BLOB NOT NULL ) """ + +# Signing records: the highest slot each validator key has signed. +# +# The signature scheme is a stateful one-time signature keyed by slot, so these +# rows are what stops a restart from consuming one slot twice. +# They are never pruned: dropping a row re-opens the key it protects. + +SIGNING_RECORDS_TABLE_NAME: Final = "signing_records" + +SIGNING_RECORDS_CREATE_TABLE: Final = """ + CREATE TABLE IF NOT EXISTS signing_records ( + validator_index INTEGER NOT NULL, + key_role TEXT NOT NULL, + last_signed_slot INTEGER NOT NULL, + PRIMARY KEY (validator_index, key_role) + ) +""" diff --git a/src/lean_spec/node/storage/sqlite.py b/src/lean_spec/node/storage/sqlite.py index 97ad14b2a..0df19c168 100644 --- a/src/lean_spec/node/storage/sqlite.py +++ b/src/lean_spec/node/storage/sqlite.py @@ -29,6 +29,8 @@ CHECKPOINTS_KEY_HEAD, CHECKPOINTS_KEY_JUSTIFIED, CHECKPOINTS_TABLE_NAME, + SIGNING_RECORDS_CREATE_TABLE, + SIGNING_RECORDS_TABLE_NAME, SLOT_INDEX_CREATE_TABLE, SLOT_INDEX_TABLE_NAME, STATE_ROOT_INDEX_CREATE_TABLE, @@ -37,7 +39,7 @@ STATES_CREATE_TABLE, STATES_TABLE_NAME, ) -from lean_spec.spec.forks import Checkpoint, Slot +from lean_spec.spec.forks import Checkpoint, Slot, ValidatorIndex from lean_spec.spec.forks.protocol import ( SpecBlockType, SpecStateType, @@ -91,6 +93,8 @@ def _init_schema(self) -> None: cursor.execute(SLOT_INDEX_CREATE_TABLE) cursor.execute(STATE_ROOT_INDEX_CREATE_TABLE) + cursor.execute(SIGNING_RECORDS_CREATE_TABLE) + self._connection.commit() # Block Operations @@ -397,6 +401,53 @@ def put_genesis_time(self, genesis_time: Uint64) -> None: except sqlite3.Error as exception: raise StorageWriteError(f"Failed to write genesis time: {exception}") from exception + # Signing Records + + def get_last_signed_slot(self, validator_index: ValidatorIndex, key_role: str) -> Slot | None: + """Retrieve the highest slot one of a validator's keys has signed.""" + try: + cursor = self._connection.cursor() + cursor.execute( + f""" + SELECT last_signed_slot FROM {SIGNING_RECORDS_TABLE_NAME} + WHERE validator_index = ? AND key_role = ? + """, + (int(validator_index), key_role), + ) + row = cursor.fetchone() + except sqlite3.Error as exception: + raise StorageReadError( + f"Failed to read signing record for validator {validator_index} " + f"{key_role} key: {exception}" + ) from exception + + if row is None: + return None + return Slot(row["last_signed_slot"]) + + def put_last_signed_slot( + self, + validator_index: ValidatorIndex, + key_role: str, + slot: Slot, + ) -> None: + """Record the highest slot one of a validator's keys has signed.""" + try: + cursor = self._connection.cursor() + cursor.execute( + f""" + INSERT OR REPLACE INTO {SIGNING_RECORDS_TABLE_NAME} + (validator_index, key_role, last_signed_slot) + VALUES (?, ?, ?) + """, + (int(validator_index), key_role, int(slot)), + ) + except sqlite3.Error as exception: + raise StorageWriteError( + f"Failed to write signing record for validator {validator_index} " + f"{key_role} key at slot {slot}: {exception}" + ) from exception + # Transaction Control @contextmanager diff --git a/src/lean_spec/node/validator/__init__.py b/src/lean_spec/node/validator/__init__.py index 4aebf1ab2..2a025ee21 100644 --- a/src/lean_spec/node/validator/__init__.py +++ b/src/lean_spec/node/validator/__init__.py @@ -8,8 +8,14 @@ from lean_spec.node.validator.registry import ValidatorRegistry from lean_spec.node.validator.service import ValidatorService +from lean_spec.node.validator.signing_protection import ( + SigningProtection, + SigningProtectionError, +) __all__ = [ "ValidatorService", "ValidatorRegistry", + "SigningProtection", + "SigningProtectionError", ] diff --git a/src/lean_spec/node/validator/service.py b/src/lean_spec/node/validator/service.py index fb0650808..3cbaaa6c6 100644 --- a/src/lean_spec/node/validator/service.py +++ b/src/lean_spec/node/validator/service.py @@ -16,6 +16,11 @@ SYNC_LAG_THRESHOLD, ) from lean_spec.node.validator.registry import ValidatorEntry, ValidatorRegistry +from lean_spec.node.validator.signing_protection import ( + SigningProtection, + SigningProtectionError, + SigningRole, +) from lean_spec.spec.crypto.merkleization import hash_tree_root from lean_spec.spec.crypto.xmss import TARGET_SIGNATURE_SCHEME from lean_spec.spec.crypto.xmss.containers import PublicKey, Signature @@ -42,6 +47,12 @@ ATTESTED_SLOT_RETENTION: Final[int] = 4 """Slots of attestation dedup history to keep; older slots can no longer be attested.""" +SIGNING_ROLE_BY_KEY_FIELD: Final[dict[str, SigningRole]] = { + "attestation_secret_key": "attestation", + "proposal_secret_key": "proposal", +} +"""Maps a registry key field to the role name signing records are stored under.""" + @dataclass(slots=True) class ValidatorService: @@ -70,6 +81,9 @@ class ValidatorService: on_attestation: AttestationPublisher | None = field(default=None) """Callback to publish a produced attestation, or None in tests and offline runs.""" + signing_protection: SigningProtection = field(default_factory=SigningProtection) + """Guard against spending one key twice in a slot. Adopts the node database if it has none.""" + _running: bool = field(default=False, repr=False) """Whether the service is running.""" @@ -85,6 +99,11 @@ class ValidatorService: _duty_gate_closed: bool = field(default=False, repr=False) """Hysteresis flag. True while signing is silenced.""" + def __post_init__(self) -> None: + """Point signing protection at the node database unless it already has a store.""" + if self.signing_protection.database is None: + self.signing_protection.database = self.sync_service.database + async def run(self) -> None: """ Check and run duties once per interval until stopped. @@ -95,6 +114,13 @@ async def run(self) -> None: self._running = True last_handled_total_interval: Interval | None = None + if not self.signing_protection.is_durable: + logger.warning( + "Signing protection is not durable: no database is configured. " + "A restart or a backward clock step within a slot can sign it twice " + "and expose one-time key material." + ) + while self._running: # Get current total interval count (not just within-slot). total_interval = self.clock.total_intervals() @@ -219,6 +245,14 @@ async def _maybe_produce_block(self, slot: Slot) -> None: slot, exception, ) + except SigningProtectionError as exception: + # The proposal key already spent this slot; signing again would expose it. + logger.warning( + "Block production skipped for validator %d at slot %d: %s", + validator_index, + slot, + exception, + ) async def _produce_attestations(self, slot: Slot) -> None: """Produce and gossip an attestation for every validator we control.""" @@ -250,15 +284,27 @@ async def _produce_attestations(self, slot: Slot) -> None: raise ValueError(f"No secret key for validator {validator_index}") attestation_data = self.spec.produce_attestation_data(store, slot) - signed_attestation = SignedAttestation( - validator_index=validator_index, - data=attestation_data, - signature=self._sign_with_key( + try: + attestation_signature = self._sign_with_key( validator_entry, attestation_data.slot, hash_tree_root(attestation_data), "attestation_secret_key", - ), + ) + except SigningProtectionError as exception: + # One validator's spent key must not silence the others. + logger.warning( + "Attestation skipped for validator %d at slot %d: %s", + validator_index, + slot, + exception, + ) + continue + + signed_attestation = SignedAttestation( + validator_index=validator_index, + data=attestation_data, + signature=attestation_signature, ) self._attestations_produced += 1 @@ -381,7 +427,18 @@ def _sign_with_key( XMSS keys are stateful one-time signatures, so each signature consumes key state. The advanced key is written back so the next slot does not reuse it. + + Raises SigningProtectionError when this key already signed the slot, which + keeps a restart or a backward clock step from opening the one-time key. """ + # Claim the slot first: a record without a signature costs a duty, + # a signature without a record costs the key. + self.signing_protection.reserve( + validator_entry.index, + SIGNING_ROLE_BY_KEY_FIELD[key_field], + slot, + ) + scheme = TARGET_SIGNATURE_SCHEME secret_key = getattr(validator_entry, key_field) diff --git a/src/lean_spec/node/validator/signing_protection.py b/src/lean_spec/node/validator/signing_protection.py new file mode 100644 index 000000000..218c66c03 --- /dev/null +++ b/src/lean_spec/node/validator/signing_protection.py @@ -0,0 +1,86 @@ +""" +Protection against consuming one validator key twice within a slot. + +The signature scheme is a stateful one-time signature indexed by slot. +Signing two different messages under the same key and slot opens two positions +in the same hash chains, which is enough to forge a third signature. +`sign` states that obligation but does not enforce it, so the enforcement lives +here: a record of the highest slot each key has signed, checked before signing. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Literal + +from lean_spec.node.storage import Database +from lean_spec.spec.forks import Slot, ValidatorIndex + +logger = logging.getLogger(__name__) + +type SigningRole = Literal["attestation", "proposal"] +"""Which of a validator's two keys a signature consumes. + +The roles hold separate keys, so a proposal and an attestation in one slot +consume one slot each rather than colliding. +""" + + +class SigningProtectionError(Exception): + """Raised when a signature would consume a slot one of the keys already spent.""" + + +@dataclass(slots=True) +class SigningProtection: + """ + Enforces at most one signature per validator key per slot. + + Records live in the node database when one is configured, so they survive a + restart, a clock that steps backwards, and a restore from backup. Without a + database they are process-local, which leaves a crash inside a slot + unprotected — `is_durable` reports which of the two is in force. + + Two instances must never share a key. Concurrent nodes holding the same key + keep separate records and can each sign the same slot. + """ + + database: Database | None = None + """Store for signing records, or None to keep them in this process only.""" + + _slots_in_memory: dict[tuple[ValidatorIndex, SigningRole], Slot] = field(default_factory=dict) + """Records used while no database is configured.""" + + @property + def is_durable(self) -> bool: + """Whether the records survive a restart of this process.""" + return self.database is not None + + def reserve(self, validator_index: ValidatorIndex, role: SigningRole, slot: Slot) -> None: + """ + Claim a slot for one of a validator's keys, to be called before it signs. + + Raises `SigningProtectionError` when that key already signed this slot or + a later one. The claim is stored before the caller signs, so an + interruption in between forfeits the duty rather than the key. + """ + last_signed = self.last_signed_slot(validator_index, role) + if last_signed is not None and slot <= last_signed: + raise SigningProtectionError( + f"validator {validator_index} {role} key last signed slot {last_signed}; " + f"signing slot {slot} would reuse a one-time key" + ) + + if self.database is None: + self._slots_in_memory[(validator_index, role)] = slot + return + + # Commit on its own so the record outlives the signature it guards. + with self.database.batch_write(): + self.database.put_last_signed_slot(validator_index, role, slot) + + def last_signed_slot(self, validator_index: ValidatorIndex, role: SigningRole) -> Slot | None: + """Return the highest slot this key has signed, or None if it never signed.""" + if self.database is None: + return self._slots_in_memory.get((validator_index, role)) + return self.database.get_last_signed_slot(validator_index, role) diff --git a/tests/node/validator/test_signing_protection.py b/tests/node/validator/test_signing_protection.py new file mode 100644 index 000000000..c31631acd --- /dev/null +++ b/tests/node/validator/test_signing_protection.py @@ -0,0 +1,294 @@ +"""Tests for protection against spending one validator key twice in a slot.""" + +from __future__ import annotations + +import logging +import tempfile +from collections.abc import Generator +from pathlib import Path + +import pytest + +from consensus_testing import MockNetworkRequester +from consensus_testing.keys import XmssKeyManager +from lean_spec.node.chain.clock import SlotClock +from lean_spec.node.storage import SQLiteDatabase +from lean_spec.node.sync.block_cache import BlockCache +from lean_spec.node.sync.peer_manager import PeerManager +from lean_spec.node.sync.service import SyncService +from lean_spec.node.validator import ( + SigningProtection, + SigningProtectionError, + ValidatorRegistry, + ValidatorService, +) +from lean_spec.node.validator.registry import ValidatorEntry +from lean_spec.node.validator.service import AttestationPublisher +from lean_spec.spec.forks import Slot, ValidatorIndex +from lean_spec.spec.forks.lstar import State, Store +from lean_spec.spec.forks.lstar.containers import Block, SignedAttestation +from lean_spec.spec.forks.lstar.spec import LstarSpec +from lean_spec.spec.ssz import Bytes32, Uint64 + +_VALIDATOR = ValidatorIndex(0) + + +@pytest.fixture +def database_path() -> Generator[Path, None, None]: + """Path to a SQLite file that outlives a single database instance.""" + with tempfile.TemporaryDirectory() as directory: + yield Path(directory) / "node.sqlite" + + +def _open_database(path: Path) -> SQLiteDatabase: + """Open the node database at a path, creating it on first use.""" + return SQLiteDatabase(path, State, Block) + + +def _make_registry(key_manager: XmssKeyManager, *indices: int) -> ValidatorRegistry: + """Build a registry holding real XMSS keys for the given validators.""" + registry = ValidatorRegistry() + for index in indices: + validator_index = ValidatorIndex(index) + keypairs = key_manager[validator_index] + registry.add( + ValidatorEntry( + index=validator_index, + attestation_secret_key=keypairs.attestation_keypair.secret_key, + proposal_secret_key=keypairs.proposal_keypair.secret_key, + ) + ) + return registry + + +def _make_service( + store: Store, + key_manager: XmssKeyManager, + database: SQLiteDatabase | None = None, + *indices: int, + on_attestation: AttestationPublisher | None = None, +) -> ValidatorService: + """Build a validator service, optionally backed by a database.""" + sync_service = SyncService( + store=store, + peer_manager=PeerManager(), + block_cache=BlockCache(), + clock=SlotClock(genesis_time=Uint64(0)), + network=MockNetworkRequester(), + database=database, + ) + return ValidatorService( + sync_service=sync_service, + clock=SlotClock(genesis_time=Uint64(0)), + registry=_make_registry(key_manager, *(indices or (0,))), + spec=LstarSpec(), + on_attestation=on_attestation, + ) + + +def _make_block(store: Store, slot: int) -> Block: + """Build a block at a slot, descending from the store head.""" + return Block( + slot=Slot(slot), + proposer_index=_VALIDATOR, + parent_root=store.head, + state_root=store.head, + body=store.blocks[store.head].body, + ) + + +class TestReserve: + """Unit tests for the slot claim itself.""" + + def test_should_allow_first_signature_when_key_is_unused(self) -> None: + """A key that never signed can claim any slot.""" + protection = SigningProtection() + + protection.reserve(_VALIDATOR, "attestation", Slot(3)) + + assert protection.last_signed_slot(_VALIDATOR, "attestation") == Slot(3) + + def test_should_reject_second_signature_when_slot_already_signed(self) -> None: + """Claiming the same slot twice raises rather than reusing the key.""" + protection = SigningProtection() + protection.reserve(_VALIDATOR, "attestation", Slot(3)) + + with pytest.raises(SigningProtectionError, match="would reuse a one-time key"): + protection.reserve(_VALIDATOR, "attestation", Slot(3)) + + def test_should_reject_signature_when_slot_moves_backwards(self) -> None: + """A backward clock step cannot re-open an already spent slot.""" + protection = SigningProtection() + protection.reserve(_VALIDATOR, "attestation", Slot(7)) + + with pytest.raises(SigningProtectionError): + protection.reserve(_VALIDATOR, "attestation", Slot(6)) + + def test_should_allow_signature_when_slot_advances(self) -> None: + """Later slots claim their own one-time key.""" + protection = SigningProtection() + protection.reserve(_VALIDATOR, "attestation", Slot(3)) + + protection.reserve(_VALIDATOR, "attestation", Slot(4)) + + assert protection.last_signed_slot(_VALIDATOR, "attestation") == Slot(4) + + def test_should_track_roles_separately(self) -> None: + """A proposal and an attestation in one slot use separate keys.""" + protection = SigningProtection() + + protection.reserve(_VALIDATOR, "proposal", Slot(3)) + protection.reserve(_VALIDATOR, "attestation", Slot(3)) + + assert protection.last_signed_slot(_VALIDATOR, "proposal") == Slot(3) + assert protection.last_signed_slot(_VALIDATOR, "attestation") == Slot(3) + + def test_should_track_validators_separately(self) -> None: + """One validator's spent slot leaves another validator's key free.""" + protection = SigningProtection() + other_validator = ValidatorIndex(1) + + protection.reserve(_VALIDATOR, "attestation", Slot(3)) + protection.reserve(other_validator, "attestation", Slot(3)) + + assert protection.last_signed_slot(other_validator, "attestation") == Slot(3) + + def test_should_report_not_durable_when_no_database_is_configured(self) -> None: + """Records kept in the process do not survive a restart.""" + assert SigningProtection().is_durable is False + + +class TestDatabaseBackedReserve: + """Records held in the node database.""" + + def test_should_report_durable_when_database_is_configured(self, database_path: Path) -> None: + """A configured database makes the records survive a restart.""" + with _open_database(database_path) as database: + assert SigningProtection(database=database).is_durable is True + + def test_should_commit_the_claim_before_returning(self, database_path: Path) -> None: + """The record is committed on its own, not left for a later batch.""" + with _open_database(database_path) as database: + SigningProtection(database=database).reserve(_VALIDATOR, "attestation", Slot(3)) + + # A fresh connection sees the row only if the write already committed. + with _open_database(database_path) as reopened: + assert reopened.get_last_signed_slot(_VALIDATOR, "attestation") == Slot(3) + + def test_should_reject_slot_reserved_by_an_earlier_database_instance( + self, database_path: Path + ) -> None: + """The claim outlives the process that made it.""" + with _open_database(database_path) as first: + SigningProtection(database=first).reserve(_VALIDATOR, "attestation", Slot(3)) + + with _open_database(database_path) as second: + with pytest.raises(SigningProtectionError): + SigningProtection(database=second).reserve(_VALIDATOR, "attestation", Slot(3)) + + +class TestServiceSigning: + """The guard as it applies at the service's signing boundary.""" + + def test_should_adopt_the_node_database( + self, keyed_store: Store, key_manager: XmssKeyManager, database_path: Path + ) -> None: + """A service whose sync layer persists gets durable protection.""" + with _open_database(database_path) as database: + service = _make_service(keyed_store, key_manager, database) + + assert service.signing_protection.is_durable is True + + def test_should_stay_in_memory_without_a_node_database( + self, keyed_store: Store, key_manager: XmssKeyManager + ) -> None: + """A service without persistence keeps process-local records.""" + service = _make_service(keyed_store, key_manager) + + assert service.signing_protection.is_durable is False + + def test_should_sign_a_block_once_per_slot( + self, keyed_store: Store, key_manager: XmssKeyManager + ) -> None: + """The first proposal for a slot signs and records the slot.""" + service = _make_service(keyed_store, key_manager) + block = _make_block(keyed_store, slot=1) + + service._sign_block(block, _VALIDATOR, []) + + assert service.signing_protection.last_signed_slot(_VALIDATOR, "proposal") == Slot(1) + + def test_should_refuse_a_second_block_for_a_signed_slot( + self, keyed_store: Store, key_manager: XmssKeyManager + ) -> None: + """Re-proposing a slot with a different block is refused, not signed.""" + service = _make_service(keyed_store, key_manager) + service._sign_block(_make_block(keyed_store, slot=1), _VALIDATOR, []) + + # A different state root gives a different block root, so a second + # signature here would open the slot's one-time key. + divergent_block = _make_block(keyed_store, slot=1) + divergent_block = Block( + slot=divergent_block.slot, + proposer_index=divergent_block.proposer_index, + parent_root=divergent_block.parent_root, + state_root=Bytes32(b"\x01" * 32), + body=divergent_block.body, + ) + + with pytest.raises(SigningProtectionError): + service._sign_block(divergent_block, _VALIDATOR, []) + + def test_should_refuse_a_slot_signed_before_a_restart( + self, keyed_store: Store, key_manager: XmssKeyManager, database_path: Path + ) -> None: + """A crash and restart inside a slot cannot sign that slot again. + + This is the case an in-memory guard misses: the replacement service has + no memory of the first signature and only the database record stops it. + """ + # Before the crash: propose at slot 1 with persistence enabled. + with _open_database(database_path) as database: + service = _make_service(keyed_store, key_manager, database) + service._sign_block(_make_block(keyed_store, slot=1), _VALIDATOR, []) + + # After the restart: a new service, a new database handle, the same slot. + with _open_database(database_path) as reopened: + restarted = _make_service(keyed_store, key_manager, reopened) + assert restarted.signing_protection.last_signed_slot(_VALIDATOR, "proposal") == Slot(1) + + with pytest.raises(SigningProtectionError): + restarted._sign_block(_make_block(keyed_store, slot=1), _VALIDATOR, []) + + +class TestDutySkipsOnRefusal: + """A refused signature skips the duty instead of failing the duty loop.""" + + async def test_should_skip_only_the_validator_whose_key_is_spent( + self, keyed_store: Store, key_manager: XmssKeyManager + ) -> None: + """One spent attestation key must not silence the other validators.""" + published: list[SignedAttestation] = [] + + async def capture(attestation: SignedAttestation) -> None: + published.append(attestation) + + service = _make_service(keyed_store, key_manager, None, 0, 1, on_attestation=capture) + service.signing_protection.reserve(_VALIDATOR, "attestation", Slot(0)) + + await service._produce_attestations(Slot(0)) + + assert [int(attestation.validator_index) for attestation in published] == [1] + + async def test_should_skip_block_production_when_proposal_key_is_spent( + self, keyed_store: Store, key_manager: XmssKeyManager, caplog: pytest.LogCaptureFixture + ) -> None: + """A spent proposal key skips the proposal rather than raising into the duty loop.""" + service = _make_service(keyed_store, key_manager) + service.signing_protection.reserve(_VALIDATOR, "proposal", Slot(8)) + + with caplog.at_level(logging.WARNING): + await service._maybe_produce_block(Slot(8)) + + assert "would reuse a one-time key" in caplog.text + assert service.blocks_produced == 0