feat(validator): refuse to sign a slot a key already signed

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

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)
+    )
+"""

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

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",
 ]

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)
 

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)