Ed25519 async batch verification for JoinSplit signatures

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: