Correct the prior documented TOCTOU

Now, if a malicious validator set publishes a malicious `Batch` at the last
moment, it'll cause all future `Batch`s signed by the next validator set to
require a bool being set (yet they never will set it).

This will prevent the handover.

The only overhead is having two distinct `batch_message` calls on-chain.

coordinator/src/main.rs

@@ -924,15 +924,6 @@ async fn handle_processor_messages<D: Db, Pro: Processors, P: P2p>(
                   MainDb::<D>::set_did_handover(&mut txn, spec.set());
                 }
 
-                // TODO: There is a race condition here. We may verify all `Batch`s from the prior
-                // set, start signing the handover Batch `n`, start signing `n+1`, have `n+1`
-                // signed before `n` (or at the same time), yet then the prior set forges a
-                // malicious Batch `n`.
-                //
-                // The malicious Batch `n` would be publishable to Serai, as Serai can't
-                // distinguish what's intended to be a handover `Batch`, yet then anyone could
-                // publish the new set's `n+1`, causing their acceptance of the handover.
-
                 Some(Transaction::Batch(block.0, id.id))
               } else {
                 Some(Transaction::BatchPreprocess(SignData {

processor/src/substrate_signer.rs

@@ -52,6 +52,18 @@ impl<D: Db> SubstrateSignerDb<D> {
     D::key(b"SUBSTRATE_SIGNER", dst, key)
   }
 
+  fn first_batch_key(key: [u8; 32]) -> Vec<u8> {
+    Self::sign_key(b"first_batch", key)
+  }
+  fn save_first_batch(txn: &mut D::Transaction<'_>, key: [u8; 32], id: u32) {
+    txn.put(Self::first_batch_key(key), id.to_le_bytes());
+  }
+  fn first_batch<G: Get>(getter: &G, key: [u8; 32]) -> Option<u32> {
+    getter
+      .get(Self::first_batch_key(key))
+      .map(|bytes| u32::from_le_bytes(bytes.try_into().unwrap()))
+  }
+
   fn completed_key(id: [u8; 32]) -> Vec<u8> {
     Self::sign_key(b"completed", id)
   }
@@ -229,6 +241,11 @@ impl<D: Db> SubstrateSigner<D> {
       return;
     }
 
+    let group_key = self.keys.group_key().to_bytes();
+    if SubstrateSignerDb::<D>::first_batch(txn, group_key).is_none() {
+      SubstrateSignerDb::<D>::save_first_batch(txn, group_key, batch.id);
+    }
+
     self.signable.insert(id, batch);
     self.attempt(txn, id, 0).await;
   }
@@ -270,8 +287,13 @@ impl<D: Db> SubstrateSigner<D> {
           Err(e) => todo!("malicious signer: {:?}", e),
         };
 
+        let batch = &self.signable[&id.id];
+        let is_first_batch =
+          SubstrateSignerDb::<D>::first_batch(txn, self.keys.group_key().to_bytes()).unwrap() ==
+            batch.id;
+
         let (machine, share) =
-          match machine.sign(preprocesses, &batch_message(&self.signable[&id.id])) {
+          match machine.sign(preprocesses, &batch_message(is_first_batch, batch)) {
             Ok(res) => res,
             Err(e) => todo!("malicious signer: {:?}", e),
           };

processor/src/tests/substrate_signer.rs

@@ -146,8 +146,9 @@ async fn test_substrate_signer() {
       signers.get_mut(i).unwrap().events.pop_front().unwrap()
     {
       assert_eq!(signed_batch.batch, batch);
+      // SubstrateSigner will believe this is the first batch for this set, hence `true`
       assert!(Public::from_raw(keys[&participant_one].group_key().to_bytes())
-        .verify(&batch_message(&batch), &signed_batch.signature));
+        .verify(&batch_message(true, &batch), &signed_batch.signature));
     } else {
       panic!("didn't get signed batch back");
     }

substrate/client/tests/common/in_instructions.rs

@@ -39,7 +39,8 @@ pub async fn provide_batch(batch: Batch) -> [u8; 32] {
 
   let block = publish_tx(&Serai::execute_batch(SignedBatch {
     batch: batch.clone(),
-    signature: pair.sign(&batch_message(&batch)),
+    // TODO: This `batch.id == 0` line only works when session == 0
+    signature: pair.sign(&batch_message(batch.id == 0, &batch)),
   }))
   .await;
 

substrate/in-instructions/pallet/src/lib.rs

@@ -155,14 +155,21 @@ pub mod pallet {
       // verify the signature
       let network = batch.batch.network;
       let (current_session, prior, current) = keys_for_network::<T>(network)?;
-      let batch_message = batch_message(&batch.batch);
       // Check the prior key first since only a single `Batch` (the last one) will be when prior is
       // Some yet prior wasn't the signing key
-      let valid_by_prior =
-        if let Some(key) = prior { key.verify(&batch_message, &batch.signature) } else { false };
+      let valid_by_prior = if let Some(key) = prior {
+        key.verify(&batch_message(false, &batch.batch), &batch.signature)
+      } else {
+        false
+      };
       let valid = valid_by_prior ||
         (if let Some(key) = current {
-          key.verify(&batch_message, &batch.signature)
+          key.verify(
+            // This `== 0` is valid as either it'll be the first Batch for the first set, or if
+            // they never had a Batch, the first Batch for the next set
+            &batch_message((batch.batch.id == 0) || prior.is_some(), &batch.batch),
+            &batch.signature,
+          )
         } else {
           false
         });

substrate/in-instructions/primitives/src/lib.rs

@@ -84,6 +84,6 @@ impl Zeroize for SignedBatch {
 
 // TODO: Make this an associated method?
 /// The message for the batch signature.
-pub fn batch_message(batch: &Batch) -> Vec<u8> {
-  [b"InInstructions-batch".as_ref(), &batch.encode()].concat()
+pub fn batch_message(is_first_batch_of_set: bool, batch: &Batch) -> Vec<u8> {
+  [b"InInstructions-batch".as_ref(), &(is_first_batch_of_set, batch).encode()].concat()
 }

tests/coordinator/src/tests/batch.rs

@@ -165,7 +165,7 @@ pub async fn batch(
   let signature = Signature(
     schnorrkel::keys::Keypair::from_bytes(&schnorrkel_key_pair)
       .unwrap()
-      .sign_simple(b"substrate", &batch_message(&batch))
+      .sign_simple(b"substrate", &batch_message(batch.id == 0, &batch))
       .to_bytes(),
   );
 

tests/processor/src/tests/batch.rs

@@ -137,8 +137,10 @@ pub(crate) async fn sign_batch(
           messages::substrate::ProcessorMessage::SignedBatch { batch: this_batch },
         ) => {
           if batch.is_none() {
-            assert!(PublicKey::from_raw(key)
-              .verify(&batch_message(&this_batch.batch), &this_batch.signature));
+            assert!(PublicKey::from_raw(key).verify(
+              &batch_message(this_batch.batch.id == 0, &this_batch.batch),
+              &this_batch.signature
+            ));
 
             batch = Some(this_batch.clone());
           }