fix(net): validate peer transactions before mempool insertion

lib/mempool.rs

@@ -180,3 +180,165 @@ impl MemPool {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod p2p_validation_bypass_tests {
+    //! Regression tests for the P2P transaction-validation gap.
+    //! MemPool::put tracks conflicts, but it is not a transaction validator.
+    //! Peer-supplied transactions must pass State::validate_transaction before
+    //! the net task regenerates proofs or inserts them into the mempool.
+
+    use bitcoin::Amount;
+    use heed::EnvOpenOptions;
+
+    use super::MemPool;
+    use crate::authorization::{Authorization, get_address, sign};
+    use crate::state::State;
+    use crate::types::{
+        Accumulator, AccumulatorDiff, Address, OutPoint, OutPointKey, Output,
+        OutputContent, PointedOutput, Transaction, Txid, hash,
+    };
+
+    // ed25519 signing keys from fixed seeds (deterministic; no RNG).
+    fn signing_key(seed: u8) -> crate::authorization::SigningKey {
+        crate::authorization::SigningKey::from_bytes(&[seed; 32])
+    }
+
+    fn temp_env() -> sneed::Env {
+        let dir = std::env::temp_dir()
+            .join(format!("thunder-m001-{}", std::process::id()));
+        drop(std::fs::remove_dir_all(&dir));
+        std::fs::create_dir_all(&dir).expect("create temp env dir");
+        let mut opts = EnvOpenOptions::new();
+        opts.map_size(16 * 1024 * 1024)
+            .max_dbs(State::NUM_DBS + MemPool::NUM_DBS + 4);
+        unsafe { sneed::Env::open(&opts, &dir) }.expect("open env")
+    }
+
+    #[test]
+    fn mempool_put_is_not_transaction_validation() {
+        let env = temp_env();
+        let state = State::new(&env).expect("State::new");
+        let mempool = MemPool::new(&env).expect("MemPool::new");
+
+        let victim = signing_key(1);
+        let victim_addr: Address = get_address(&victim.verifying_key());
+        let funding_outpoint = OutPoint::Regular {
+            txid: Txid([7u8; 32]),
+            vout: 0,
+        };
+        let funded_output = Output {
+            address: victim_addr,
+            content: OutputContent::Value(Amount::from_sat(100_000)),
+        };
+        let utxo_hash = hash(&PointedOutput {
+            outpoint: funding_outpoint,
+            output: funded_output.clone(),
+        });
+        {
+            let mut rwtxn = env.write_txn().expect("write txn");
+            state
+                .utxos
+                .put(
+                    &mut rwtxn,
+                    &OutPointKey::from(funding_outpoint),
+                    &funded_output,
+                )
+                .expect("put utxo");
+            // Keep the accumulator consistent so proof regeneration succeeds.
+            let mut acc: Accumulator =
+                state.get_accumulator(&rwtxn).expect("get accumulator");
+            let mut diff = AccumulatorDiff::default();
+            diff.insert(utxo_hash.into());
+            acc.apply_diff(diff).expect("apply diff");
+            state
+                .utreexo_accumulator
+                .put(&mut rwtxn, &(), &acc)
+                .expect("put accumulator");
+            rwtxn.commit().expect("commit funding");
+        }
+
+        let tx = Transaction {
+            inputs: vec![(funding_outpoint, utxo_hash)],
+            outputs: vec![Output {
+                address: get_address(&signing_key(2).verifying_key()),
+                content: OutputContent::Value(Amount::from_sat(90_000)),
+            }],
+            ..Default::default()
+        };
+
+        {
+            let mut wrong_hash_tx = tx.clone();
+            wrong_hash_tx.inputs[0].1 = [9u8; 32];
+            let wrong_hash_authd_tx = crate::types::AuthorizedTransaction {
+                transaction: wrong_hash_tx.clone(),
+                authorizations: vec![Authorization {
+                    verifying_key: victim.verifying_key(),
+                    signature: sign(&victim, &wrong_hash_tx).expect("sign"),
+                }],
+            };
+            let rotxn = env.read_txn().expect("read txn");
+            let result =
+                state.validate_transaction(&rotxn, &wrong_hash_authd_tx);
+            assert!(
+                matches!(
+                    result,
+                    Err(crate::state::ValidateTransaction::WrongUtxoHash)
+                ),
+                "validate_transaction must reject a mismatched UTXO hash, got {result:?}"
+            );
+        }
+
+        // Present the victim verifying key, but sign with a different key.
+        let attacker = signing_key(2);
+        let forged = Authorization {
+            verifying_key: victim.verifying_key(),
+            signature: sign(&attacker, &tx).expect("sign"),
+        };
+        let mut authd_tx = crate::types::AuthorizedTransaction {
+            transaction: tx,
+            authorizations: vec![forged],
+        };
+
+        // The validator rejects the forged signature.
+        {
+            let rotxn = env.read_txn().expect("read txn");
+            let result = state.validate_transaction(&rotxn, &authd_tx);
+            assert!(
+                matches!(
+                    result,
+                    Err(crate::state::ValidateTransaction::Authorization)
+                ),
+                "validate_transaction must reject the forged-signature tx with an Authorization error, got {result:?}"
+            );
+        }
+
+        // The old lower-level sequence accepted it because proof regeneration
+        // and mempool insertion are not signature-validation gates.
+        {
+            let mut rwtxn = env.write_txn().expect("write txn");
+            state
+                .regenerate_proof(&rwtxn, &mut authd_tx.transaction)
+                .expect("regenerate_proof accepted the forged tx");
+            mempool
+                .put(&mut rwtxn, &authd_tx)
+                .expect("mempool.put accepted the invalid tx (the bug)");
+            rwtxn.commit().expect("commit mempool");
+        }
+
+        // Confirm the invalid tx is now resident in the mempool.
+        {
+            let rotxn = env.read_txn().expect("read txn");
+            let in_mempool = mempool.take_all(&rotxn).expect("take_all");
+            assert_eq!(
+                in_mempool.len(),
+                1,
+                "the forged-signature tx must be sitting in the mempool"
+            );
+            assert_eq!(
+                in_mempool[0].transaction.txid(),
+                authd_tx.transaction.txid(),
+            );
+        }
+    }
+}

lib/net/peer/task.rs

@@ -6,6 +6,7 @@ use std::{
     sync::{Arc, atomic::AtomicBool},
 };
 
+use error_fatality::{Nested as _, Split};
 use fallible_iterator::FallibleIterator;
 use futures::{StreamExt as _, channel::mpsc};
 use quinn::SendStream;
@@ -23,6 +24,7 @@ use crate::{
         message::{self, Heartbeat, RequestMessage, ResponseMessage},
         request_queue,
     },
+    state,
     types::{
         AuthorizedTransaction, BlockHash, BmmResult, Header, Tip, VERSION,
     },
@@ -659,26 +661,36 @@ impl ConnectionTask {
             let rotxn = ctxt.env.read_txn().map_err(EnvError::from)?;
             ctxt.state.validate_transaction(&rotxn, &tx)
         };
-        match validate_tx_result {
-            Err(err) => {
+        match validate_tx_result
+            .into_nested()
+            .map_err(state::Error::from)?
+        {
+            Ok(_) => {
                 Connection::send_response(
                     ctxt.network,
                     response_tx,
-                    ResponseMessage::TransactionRejected(txid),
+                    ResponseMessage::TransactionAccepted(txid),
                 )
                 .await?;
-                Err(Error::from(err))
+                info_tx
+                    .unbounded_send(Info::NewTransaction(tx))
+                    .map_err(|_| Error::SendInfo)?;
+                Ok(())
             }
-            Ok(_) => {
+            Err(non_fatal) => {
+                let non_fatal: <state::ValidateTransaction as Split>::Jfyi =
+                    non_fatal;
+                let non_fatal = anyhow::Error::from(non_fatal);
+                tracing::warn!(
+                    "Rejected invalid pushed transaction from peer: \
+                     {non_fatal:#}"
+                );
                 Connection::send_response(
                     ctxt.network,
                     response_tx,
-                    ResponseMessage::TransactionAccepted(txid),
+                    ResponseMessage::TransactionRejected(txid),
                 )
                 .await?;
-                info_tx
-                    .unbounded_send(Info::NewTransaction(tx))
-                    .map_err(|_| Error::SendInfo)?;
                 Ok(())
             }
         }

lib/node/mod.rs

@@ -249,7 +249,9 @@ where
     ) -> Result<(), Error> {
         {
             let mut rotxn = self.env.write_txn().map_err(EnvError::from)?;
-            self.state.validate_transaction(&rotxn, &transaction)?;
+            self.state
+                .validate_transaction(&rotxn, &transaction)
+                .map_err(state::Error::from)?;
             self.mempool.put(&mut rotxn, &transaction)?;
             rotxn.commit().map_err(RwTxnError::from)?;
         }
@@ -460,7 +462,8 @@ where
             }
             let filled_transaction = self
                 .state
-                .fill_authorized_transaction(&rwtxn, transaction)?;
+                .fill_authorized_transaction(&rwtxn, transaction)
+                .map_err(state::Error::from)?;
             let value_in: bitcoin::Amount = filled_transaction
                 .transaction
                 .spent_utxos

lib/node/net_task.rs

@@ -1089,11 +1089,55 @@ impl NetTask {
                                 .env
                                 .write_txn()
                                 .map_err(EnvError::from)?;
+                            // Validate the peer-supplied transaction before
+                            // accepting it into the mempool, mirroring the RPC
+                            // path (`Node::submit_transaction`). Without this,
+                            // transactions with invalid signatures / failing
+                            // balance checks are accepted and re-broadcast.
+                            // Validation failures are non-fatal: an individual
+                            // peer may send an invalid transaction, but that
+                            // must not stop the networking task.
+                            match self
+                                .ctxt
+                                .state
+                                .validate_transaction(&rwtxn, &new_tx)
+                                .into_nested()
+                                .map_err(state::Error::from)?
+                            {
+                                Ok(_) => {}
+                                Err(non_fatal) => {
+                                    let non_fatal: <state::ValidateTransaction as Split>::Jfyi =
+                                        non_fatal;
+                                    let non_fatal =
+                                        anyhow::Error::from(non_fatal);
+                                    tracing::warn!(
+                                        "Rejected invalid peer transaction: \
+                                         {non_fatal:#}"
+                                    );
+                                    continue;
+                                }
+                            }
+                            // Validation passed; regenerate the Utreexo proof
+                            // before insertion.
                             let () = self.ctxt.state.regenerate_proof(
                                 &rwtxn,
                                 &mut new_tx.transaction,
                             )?;
-                            self.ctxt.mempool.put(&mut rwtxn, &new_tx)?;
+                            // Insert into the mempool. A double-spend against
+                            // the mempool is a normal peer condition, not a
+                            // fatal error.
+                            match self.ctxt.mempool.put(&mut rwtxn, &new_tx) {
+                                Ok(()) => {}
+                                Err(mempool::Error::UtxoDoubleSpent) => {
+                                    tracing::warn!(
+                                        txid = %new_tx.transaction.txid(),
+                                        "Rejected peer transaction: UTXO \
+                                         already spent in mempool"
+                                    );
+                                    continue;
+                                }
+                                Err(err) => return Err(Error::MemPool(err)),
+                            }
                             rwtxn.commit().map_err(RwTxnError::from)?;
                             // broadcast
                             let () = self

lib/state/error.rs

@@ -1,3 +1,4 @@
+use error_fatality::{Fatality, Split};
 use sneed::{db::error as db, env::error as env, rwtxn::error as rwtxn};
 use thiserror::Error;
 use transitive::Transitive;
@@ -14,6 +15,73 @@ pub struct NoUtxo {
     pub outpoint: OutPoint,
 }
 
+/// Error returned by [`crate::state::State::validate_transaction`].
+/// Validation failures (bad signature, missing UTXO, etc.) are non-fatal:
+/// a peer can supply an invalid transaction and the node must reject it
+/// without terminating the networking task. Database errors are fatal.
+#[allow(clippy::duplicated_attributes)]
+#[derive(Debug, Error, Fatality, Split, Transitive)]
+#[split(attrs(derive(Debug, Error)))]
+#[transitive(from(db::TryGet, db::Error))]
+#[transitive(from(db::Error, sneed::Error))]
+#[transitive(from(env::Error, sneed::Error))]
+#[transitive(from(rwtxn::Error, sneed::Error))]
+pub enum ValidateTransaction {
+    #[error("failed to verify authorization")]
+    #[fatal(false)]
+    Authorization,
+    #[error(transparent)]
+    #[fatal(false)]
+    AmountOverflow(#[from] AmountOverflowError),
+    #[error(transparent)]
+    #[fatal(false)]
+    AmountUnderflow(#[from] AmountUnderflowError),
+    #[error("value in is less than value out")]
+    #[fatal(false)]
+    NotEnoughValueIn,
+    #[error(transparent)]
+    #[fatal(false)]
+    NoUtxo(#[from] NoUtxo),
+    #[error("wrong UTXO hash")]
+    #[fatal(false)]
+    WrongUtxoHash,
+    #[error("wrong public key for address")]
+    #[fatal(false)]
+    WrongPubKeyForAddress,
+    #[error(transparent)]
+    #[fatal(true)]
+    Db(#[from] sneed::Error),
+}
+
+impl From<ValidateTransaction> for Error {
+    fn from(err: ValidateTransaction) -> Self {
+        match err {
+            ValidateTransaction::Authorization => Self::Authorization,
+            ValidateTransaction::AmountOverflow(err) => {
+                Self::AmountOverflow(err)
+            }
+            ValidateTransaction::AmountUnderflow(err) => {
+                Self::AmountUnderflow(err)
+            }
+            ValidateTransaction::NotEnoughValueIn => Self::NotEnoughValueIn,
+            ValidateTransaction::NoUtxo(err) => Self::NoUtxo(err),
+            ValidateTransaction::WrongUtxoHash => Self::WrongUtxoHash,
+            ValidateTransaction::WrongPubKeyForAddress => {
+                Self::WrongPubKeyForAddress
+            }
+            ValidateTransaction::Db(err) => Self::Db(err),
+        }
+    }
+}
+
+impl From<FatalValidateTransaction> for Error {
+    fn from(err: FatalValidateTransaction) -> Self {
+        match err {
+            FatalValidateTransaction::Db(err) => Self::Db(err),
+        }
+    }
+}
+
 #[derive(Debug, Error)]
 #[error("pending withdrawal bundle {0} unknown in withdrawal_bundles")]
 #[repr(transparent)]
@@ -170,6 +238,8 @@ pub enum Error {
     Utreexo(#[from] UtreexoError),
     #[error("Utreexo proof verification failed for tx {txid}")]
     UtreexoProofFailed { txid: Txid },
+    #[error("wrong UTXO hash")]
+    WrongUtxoHash,
     #[error("Computed Utreexo roots do not match the header roots")]
     UtreexoRootsMismatch,
     #[error("utxo double spent")]