Ed25519 async batch verification for JoinSplit signatures (#1952)

* Ed25519 async batch verification for JoinSplit signatures

We've been verifying JoinSplitSigs one-by-one pre-ZIP-215. Now as we're post-ZIP-215,
we can take advantage of the batch math to validate this signatures.

I would have pumped all the joinsplits in our MAINNET_BLOCKS test vectors but these
signatures are over the sighash, which needs the NU code to compute, and once we're
doing all that set up, we're basically doing transaction validation, so.

Resolves #1944

* Repoint to latest ed25519-zebra commit with note to point at 3.0 when released

Co-authored-by: Alfredo Garcia <[email protected]>
Co-authored-by: teor <[email protected]>

zebra-chain/Cargo.toml

@@ -38,8 +38,10 @@ proptest = { version = "0.10", optional = true }
 proptest-derive = { version = "0.3.0", optional = true }
 
 # ZF deps
+
 displaydoc = "0.2.1"
-ed25519-zebra = "2"
+# TODO: upgrade ed25510-zebra to 3 when released: https://github.com/ZcashFoundation/ed25519-zebra/issues/45
+ed25519-zebra = {git = "https://github.com/ZcashFoundation/ed25519-zebra", rev = "539fad040c443302775b0f508e616418825e6c22"}
 equihash = "0.1"
 #redjubjub = "0.2"
 redjubjub = {git = "https://github.com/ZcashFoundation/redjubjub", rev = "8101eaff1cb2fca45334f77a65caa4c46e3d545b"}

zebra-consensus/src/primitives.rs

@@ -1,5 +1,6 @@
 //! Asynchronous verification of cryptographic primitives.
 
+pub mod ed25519;
 pub mod groth16;
 pub mod redjubjub;
 

zebra-consensus/src/primitives/ed25519.rs

@@ -0,0 +1,129 @@
+//! Async Ed25519 batch verifier service
+
+#[cfg(test)]
+mod tests;
+
+use std::{
+    future::Future,
+    mem,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures::future::{ready, Ready};
+use once_cell::sync::Lazy;
+use rand::thread_rng;
+
+use tokio::sync::broadcast::{channel, error::RecvError, Sender};
+use tower::{util::ServiceFn, Service};
+use tower_batch::{Batch, BatchControl};
+use tower_fallback::Fallback;
+use zebra_chain::primitives::ed25519::{batch, *};
+
+/// Global batch verification context for Ed25519 signatures.
+///
+/// This service transparently batches contemporaneous signature verifications,
+/// handling batch failures by falling back to individual verification.
+///
+/// Note that making a `Service` call requires mutable access to the service, so
+/// you should call `.clone()` on the global handle to create a local, mutable
+/// handle.
+pub static VERIFIER: Lazy<
+    Fallback<Batch<Verifier, Item>, ServiceFn<fn(Item) -> Ready<Result<(), Error>>>>,
+> = Lazy::new(|| {
+    Fallback::new(
+        Batch::new(
+            Verifier::default(),
+            super::MAX_BATCH_SIZE,
+            super::MAX_BATCH_LATENCY,
+        ),
+        // We want to fallback to individual verification if batch verification
+        // fails, so we need a Service to use. The obvious way to do this would
+        // be to write a closure that returns an async block. But because we
+        // have to specify the type of a static, we need to be able to write the
+        // type of the closure and its return value, and both closures and async
+        // blocks have eldritch types whose names cannot be written. So instead,
+        // we use a Ready to avoid an async block and cast the closure to a
+        // function (which is possible because it doesn't capture any state).
+        tower::service_fn((|item: Item| ready(item.verify_single())) as fn(_) -> _),
+    )
+});
+
+/// Ed25519 signature verifier service
+pub struct Verifier {
+    batch: batch::Verifier,
+    // This uses a "broadcast" channel, which is an mpmc channel. Tokio also
+    // provides a spmc channel, "watch", but it only keeps the latest value, so
+    // using it would require thinking through whether it was possible for
+    // results from one batch to be mixed with another.
+    tx: Sender<Result<(), Error>>,
+}
+
+impl Default for Verifier {
+    fn default() -> Self {
+        let batch = batch::Verifier::default();
+        let (tx, _) = channel(super::BROADCAST_BUFFER_SIZE);
+        Self { batch, tx }
+    }
+}
+
+/// Type alias to clarify that this `batch::Item` is a `Ed25519Item`
+pub type Item = batch::Item;
+
+impl Service<BatchControl<Item>> for Verifier {
+    type Response = ();
+    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: BatchControl<Item>) -> Self::Future {
+        match req {
+            BatchControl::Item(item) => {
+                tracing::trace!("got ed25519 item");
+                self.batch.queue(item);
+                let mut rx = self.tx.subscribe();
+                Box::pin(async move {
+                    match rx.recv().await {
+                        Ok(result) => {
+                            if result.is_ok() {
+                                tracing::trace!(?result, "validated ed25519 signature");
+                                metrics::counter!("signatures.ed25519.validated", 1);
+                            } else {
+                                tracing::trace!(?result, "invalid ed25519 signature");
+                                metrics::counter!("signatures.ed25519.invalid", 1);
+                            }
+                            result
+                        }
+                        Err(RecvError::Lagged(_)) => {
+                            tracing::error!(
+                                "ed25519 batch verification receiver lagged and lost verification results"
+                            );
+                            Err(Error::InvalidSignature)
+                        }
+                        Err(RecvError::Closed) => {
+                            panic!("ed25519 verifier was dropped without flushing")
+                        }
+                    }
+                })
+            }
+
+            BatchControl::Flush => {
+                tracing::trace!("got ed25519 flush command");
+                let batch = mem::take(&mut self.batch);
+                let _ = self.tx.send(batch.verify(thread_rng()));
+                Box::pin(async { Ok(()) })
+            }
+        }
+    }
+}
+
+impl Drop for Verifier {
+    fn drop(&mut self) {
+        // We need to flush the current batch in case there are still any pending futures.
+        let batch = mem::take(&mut self.batch);
+        let _ = self.tx.send(batch.verify(thread_rng()));
+    }
+}

zebra-consensus/src/primitives/ed25519/tests.rs

@@ -0,0 +1,72 @@
+//! Tests for Ed25519 signature verification
+
+use std::time::Duration;
+
+use color_eyre::eyre::{eyre, Result};
+use futures::stream::{FuturesUnordered, StreamExt};
+use tower::ServiceExt;
+use tower_batch::Batch;
+
+use crate::primitives::ed25519::*;
+
+async fn sign_and_verify<V>(mut verifier: V, n: usize) -> Result<(), V::Error>
+where
+    V: Service<Item, Response = ()>,
+{
+    let mut rng = thread_rng();
+    let mut results = FuturesUnordered::new();
+    for i in 0..n {
+        let span = tracing::trace_span!("sig", i);
+        let msg = b"BatchVerifyTest";
+
+        let sk = SigningKey::new(&mut rng);
+        let vk = VerificationKey::from(&sk);
+        let sig = sk.sign(&msg[..]);
+        verifier.ready_and().await?;
+        results.push(span.in_scope(|| verifier.call((vk.into(), sig, msg).into())))
+    }
+
+    while let Some(result) = results.next().await {
+        result?;
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn batch_flushes_on_max_items_test() -> Result<()> {
+    batch_flushes_on_max_items().await
+}
+
+#[spandoc::spandoc]
+async fn batch_flushes_on_max_items() -> Result<()> {
+    use tokio::time::timeout;
+
+    // Use a very long max_latency and a short timeout to check that
+    // flushing is happening based on hitting max_items.
+    let verifier = Batch::new(Verifier::default(), 10, Duration::from_secs(1000));
+    timeout(Duration::from_secs(5), sign_and_verify(verifier, 100))
+        .await?
+        .map_err(|e| eyre!(e))?;
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn batch_flushes_on_max_latency_test() -> Result<()> {
+    batch_flushes_on_max_latency().await
+}
+
+#[spandoc::spandoc]
+async fn batch_flushes_on_max_latency() -> Result<()> {
+    use tokio::time::timeout;
+
+    // Use a very high max_items and a short timeout to check that
+    // flushing is happening based on hitting max_latency.
+    let verifier = Batch::new(Verifier::default(), 100, Duration::from_millis(500));
+    timeout(Duration::from_secs(5), sign_and_verify(verifier, 10))
+        .await?
+        .map_err(|e| eyre!(e))?;
+
+    Ok(())
+}

zebra-consensus/src/primitives/groth16.rs

@@ -175,7 +175,17 @@ impl Service<BatchControl<Item>> for Verifier {
                 let mut rx = self.tx.subscribe();
                 Box::pin(async move {
                     match rx.recv().await {
-                        Ok(result) => result,
+                        Ok(result) => {
+                            if result.is_ok() {
+                                tracing::trace!(?result, "verified groth16 proof");
+                                metrics::counter!("proofs.groth16.verified", 1);
+                            } else {
+                                tracing::trace!(?result, "invalid groth16 proof");
+                                metrics::counter!("proofs.groth16.invalid", 1);
+                            }
+
+                            result
+                        }
                         Err(RecvError::Lagged(_)) => {
                             tracing::error!(
                                 "missed channel updates, BROADCAST_BUFFER_SIZE is too low!!"

zebra-consensus/src/primitives/groth16/tests.rs

@@ -5,9 +5,7 @@ use tower::ServiceExt;
 
 use zebra_chain::{block::Block, serialization::ZcashDeserializeInto, transaction::Transaction};
 
-use crate::primitives::groth16;
-
-use super::*;
+use crate::primitives::groth16::{self, *};
 
 async fn verify_groth16_spends_and_outputs<V>(
     spend_verifier: &mut V,

zebra-consensus/src/primitives/redjubjub.rs

@@ -87,7 +87,17 @@ impl Service<BatchControl<Item>> for Verifier {
                 let mut rx = self.tx.subscribe();
                 Box::pin(async move {
                     match rx.recv().await {
-                        Ok(result) => result,
+                        Ok(result) => {
+                            if result.is_ok() {
+                                tracing::trace!(?result, "validated redjubjub signature");
+                                metrics::counter!("signatures.redjubjub.validated", 1);
+                            } else {
+                                tracing::trace!(?result, "invalid redjubjub signature");
+                                metrics::counter!("signatures.redjubjub.invalid", 1);
+                            }
+
+                            result
+                        }
                         Err(RecvError::Lagged(_)) => {
                             tracing::error!(
                                 "batch verification receiver lagged and lost verification results"

zebra-consensus/src/transaction.rs

@@ -129,6 +129,7 @@ where
         let mut spend_verifier = primitives::groth16::SPEND_VERIFIER.clone();
         let mut output_verifier = primitives::groth16::OUTPUT_VERIFIER.clone();
 
+        let mut ed25519_verifier = primitives::ed25519::VERIFIER.clone();
         let mut redjubjub_verifier = primitives::redjubjub::VERIFIER.clone();
         let mut script_verifier = self.script_verifier.clone();
 
@@ -192,7 +193,24 @@ where
                         // correctly.
 
                         // Then, pass those items to self.joinsplit to verify them.
-                        check::validate_joinsplit_sig(joinsplit_data, shielded_sighash.as_bytes())?;
+
+                        // Consensus rule: The joinSplitSig MUST represent a
+                        // valid signature, under joinSplitPubKey, of the
+                        // sighash.
+                        //
+                        // Queue the validation of the JoinSplit signature while
+                        // adding the resulting future to our collection of
+                        // async checks that (at a minimum) must pass for the
+                        // transaction to verify.
+                        //
+                        // https://zips.z.cash/protocol/protocol.pdf#sproutnonmalleability
+                        // https://zips.z.cash/protocol/protocol.pdf#txnencodingandconsensus
+                        let rsp = ed25519_verifier
+                            .ready_and()
+                            .await?
+                            .call((joinsplit_data.pub_key, joinsplit_data.sig, &shielded_sighash).into());
+
+                        async_checks.push(rsp.boxed());
                     }
 
                     if let Some(shielded_data) = shielded_data {

zebra-consensus/src/transaction/check.rs

@@ -2,31 +2,14 @@
 //!
 //! Code in this file can freely assume that no pre-V4 transactions are present.
 
-use std::convert::TryFrom;
-
 use zebra_chain::{
     amount::Amount,
-    primitives::{ed25519, Groth16Proof},
     sapling::{Output, Spend},
-    transaction::{JoinSplitData, ShieldedData, Transaction},
+    transaction::{ShieldedData, Transaction},
 };
 
 use crate::error::TransactionError;
 
-/// Validate the JoinSplit binding signature.
-///
-/// https://zips.z.cash/protocol/protocol.pdf#sproutnonmalleability
-/// https://zips.z.cash/protocol/protocol.pdf#txnencodingandconsensus
-pub fn validate_joinsplit_sig(
-    joinsplit_data: &JoinSplitData<Groth16Proof>,
-    sighash: &[u8],
-) -> Result<(), TransactionError> {
-    // TODO: batch verify ed25519: https://github.com/ZcashFoundation/zebra/issues/1944
-    ed25519::VerificationKey::try_from(joinsplit_data.pub_key)
-        .and_then(|vk| vk.verify(&joinsplit_data.sig, sighash))
-        .map_err(TransactionError::Ed25519)
-}
-
 /// Checks that the transaction has inputs and outputs.
 ///
 /// More specifically: