iota_core/

signature_verifier.rs

// Copyright (c) Mysten Labs, Inc.
// Modifications Copyright (c) 2024 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use std::{collections::BTreeSet, sync::Arc};

use either::Either;
use fastcrypto::traits::{AggregateAuthenticator, ToFromBytes};
use futures::pin_mut;
use iota_metrics::monitored_scope;
use iota_sdk_types::{
    CertificateDigest, SenderSignedDataDigest, SenderSignedTransaction, crypto::Intent,
};
use iota_types::{
    base_types::AuthorityName,
    committee::Committee,
    crypto::{AuthorityPublicKey, AuthoritySignInfoTrait, VerificationObligation},
    error::{IotaError, IotaResult},
    message_envelope::Message,
    messages_checkpoint::{CheckpointSummaryExt, SignedCheckpointSummary},
    messages_consensus::{AuthorityCapabilitiesDigest, SignedAuthorityCapabilitiesV1},
    signature::VerifyParams,
    signature_verification::{VerifiedDigestCache, verify_sender_signed_data_message_signatures},
    transaction::{CertifiedTransaction, VerifiedCertificate},
};
use itertools::{Itertools as _, izip};
use parking_lot::{Mutex, MutexGuard};
use prometheus_filtered::{IntCounter, Registry, register_int_counter_with_registry};
use tap::TapFallible;
use tokio::{
    runtime::Handle,
    sync::oneshot,
    time::{Duration, timeout},
};
use tracing::{Instrument, instrument, trace_span};
// Maximum amount of time we wait for a batch to fill up before verifying a
// partial batch.
const BATCH_TIMEOUT_MS: Duration = Duration::from_millis(10);

// Maximum size of batch to verify. Increasing this value will slightly improve
// CPU utilization (batching starts to hit steeply diminishing marginal returns
// around batch sizes of 16), at the cost of slightly increasing latency
// (BATCH_TIMEOUT_MS will be hit more frequently if system is not heavily
// loaded).
const MAX_BATCH_SIZE: usize = 8;

type Sender = oneshot::Sender<IotaResult<VerifiedCertificate>>;

struct CertBuffer {
    certs: Vec<CertifiedTransaction>,
    senders: Vec<Sender>,
    id: u64,
}

impl CertBuffer {
    fn new(capacity: usize) -> Self {
        Self {
            certs: Vec::with_capacity(capacity),
            senders: Vec::with_capacity(capacity),
            id: 0,
        }
    }

    // Function consumes MutexGuard, therefore releasing the lock after mem swap is
    // done
    fn take_and_replace(mut guard: MutexGuard<'_, Self>) -> Self {
        let this = &mut *guard;
        let mut new = CertBuffer::new(this.capacity());
        new.id = this.id + 1;
        std::mem::swap(&mut new, this);
        new
    }

    fn capacity(&self) -> usize {
        debug_assert_eq!(self.certs.capacity(), self.senders.capacity());
        self.certs.capacity()
    }

    fn len(&self) -> usize {
        debug_assert_eq!(self.certs.len(), self.senders.len());
        self.certs.len()
    }

    fn push(&mut self, tx: Sender, cert: CertifiedTransaction) {
        self.senders.push(tx);
        self.certs.push(cert);
    }
}

/// Verifies signatures in ways that faster than verifying each signature
/// individually.
/// - BLS signatures - caching and batch verification.
/// - User signed data - caching.
pub struct SignatureVerifier {
    committee: Arc<Committee>,
    non_committee_validators: BTreeSet<AuthorityName>,

    certificate_cache: VerifiedDigestCache<CertificateDigest>,
    signed_data_cache: VerifiedDigestCache<SenderSignedDataDigest>,
    authority_capability_cache: VerifiedDigestCache<AuthorityCapabilitiesDigest>,

    /// Params for signature verification.
    verify_params: VerifyParams,

    queue: Mutex<CertBuffer>,
    pub metrics: Arc<SignatureVerifierMetrics>,
}

impl SignatureVerifier {
    pub fn new_with_batch_size(
        committee: Arc<Committee>,
        non_committee_validators: BTreeSet<AuthorityName>,
        batch_size: usize,
        metrics: Arc<SignatureVerifierMetrics>,
        accept_passkey_in_multisig: bool,
        additional_multisig_checks: bool,
    ) -> Self {
        Self {
            committee,
            non_committee_validators,
            certificate_cache: VerifiedDigestCache::new(
                metrics.certificate_signatures_cache_hits.clone(),
                metrics.certificate_signatures_cache_misses.clone(),
                metrics.certificate_signatures_cache_evictions.clone(),
            ),
            signed_data_cache: VerifiedDigestCache::new(
                metrics.signed_data_cache_hits.clone(),
                metrics.signed_data_cache_misses.clone(),
                metrics.signed_data_cache_evictions.clone(),
            ),
            authority_capability_cache: VerifiedDigestCache::new(
                metrics.authority_capabilities_cache_hits.clone(),
                metrics.authority_capabilities_cache_misses.clone(),
                metrics.authority_capabilities_cache_evictions.clone(),
            ),
            queue: Mutex::new(CertBuffer::new(batch_size)),
            metrics,
            verify_params: VerifyParams::new(
                accept_passkey_in_multisig,
                additional_multisig_checks,
            ),
        }
    }

    pub fn new(
        committee: Arc<Committee>,
        non_committee_validators: BTreeSet<AuthorityName>,
        metrics: Arc<SignatureVerifierMetrics>,
        accept_passkey_in_multisig: bool,
        additional_multisig_checks: bool,
    ) -> Self {
        Self::new_with_batch_size(
            committee,
            non_committee_validators,
            MAX_BATCH_SIZE,
            metrics,
            accept_passkey_in_multisig,
            additional_multisig_checks,
        )
    }

    /// Verifies all certs, returns Ok only if all are valid.
    #[instrument(level = "trace", skip_all)]
    pub fn verify_certs_and_checkpoints(
        &self,
        certs: Vec<&CertifiedTransaction>,
        checkpoints: Vec<&SignedCheckpointSummary>,
        authority_capabilities: Vec<&SignedAuthorityCapabilitiesV1>,
    ) -> IotaResult {
        // Verify all user sigs, since caching is handled by the underlying
        // implementation.
        for cert in &certs {
            self.verify_tx(cert.data())?;
        }

        // Verify authority capabilities signatures. Caching is handled inside to avoid
        // checking the same message multiple times.
        for cap in &authority_capabilities {
            self.verify_authority_capabilities(cap)?;
        }

        batch_verify_all_certificates_and_checkpoints(&self.committee, &certs, &checkpoints)?;
        Ok(())
    }

    /// Verifies one cert asynchronously, in a batch.
    pub async fn verify_cert(&self, cert: CertifiedTransaction) -> IotaResult<VerifiedCertificate> {
        let cert_digest = cert.certificate_digest();
        if self.certificate_cache.is_cached(&cert_digest) {
            return Ok(VerifiedCertificate::new_unchecked(cert));
        }
        self.verify_tx(cert.data())?;
        self.verify_cert_skip_cache(cert)
            .await
            .tap_ok(|_| self.certificate_cache.cache_digest(cert_digest))
    }

    pub async fn multi_verify_certs(
        &self,
        certs: Vec<CertifiedTransaction>,
    ) -> Vec<IotaResult<VerifiedCertificate>> {
        // TODO: We could do better by pushing the all of `certs` into the verification
        // queue at once, but that's significantly more complex.
        let mut futures = Vec::with_capacity(certs.len());
        for cert in certs {
            futures.push(self.verify_cert(cert));
        }
        futures::future::join_all(futures).await
    }

    /// exposed as a public method for the benchmarks
    pub async fn verify_cert_skip_cache(
        &self,
        cert: CertifiedTransaction,
    ) -> IotaResult<VerifiedCertificate> {
        // this is the only innocent error we are likely to encounter - filter it before
        // we poison a whole batch.
        if cert.auth_sig().epoch != self.committee.epoch() {
            return Err(IotaError::WrongEpoch {
                expected_epoch: self.committee.epoch(),
                actual_epoch: cert.auth_sig().epoch,
            });
        }

        self.verify_cert_inner(cert).await
    }

    async fn verify_cert_inner(
        &self,
        cert: CertifiedTransaction,
    ) -> IotaResult<VerifiedCertificate> {
        // Cancellation safety: we use parking_lot locks, which cannot be held across
        // awaits. Therefore once the queue has been taken by a thread, it is
        // guaranteed to process the queue and send all results before the
        // future can be cancelled by the caller.
        let (tx, rx) = oneshot::channel();
        pin_mut!(rx);

        let prev_id_or_buffer = {
            let mut queue = self.queue.lock();
            queue.push(tx, cert);
            if queue.len() == queue.capacity() {
                Either::Right(CertBuffer::take_and_replace(queue))
            } else {
                Either::Left(queue.id)
            }
        };
        let prev_id = match prev_id_or_buffer {
            Either::Left(prev_id) => prev_id,
            Either::Right(buffer) => {
                self.metrics.full_batches.inc();
                self.process_queue(buffer)
                    .instrument(trace_span!("SignatureVerifier::process_queue"))
                    .await;
                // unwrap ok - process_queue will have sent the result already
                return rx.try_recv().unwrap();
            }
        };

        if let Ok(res) = timeout(BATCH_TIMEOUT_MS, &mut rx).await {
            // unwrap ok - tx cannot have been dropped without sending a result.
            return res.unwrap();
        }
        self.metrics.timeouts.inc();

        let buffer = {
            let queue = self.queue.lock();
            // check if another thread took the queue while we were re-acquiring lock.
            if prev_id == queue.id {
                debug_assert_ne!(queue.len(), queue.capacity());
                Some(CertBuffer::take_and_replace(queue))
            } else {
                None
            }
        };

        if let Some(buffer) = buffer {
            self.metrics.partial_batches.inc();
            self.process_queue(buffer).await;
            // unwrap ok - process_queue will have sent the result already
            return rx.try_recv().unwrap();
        }

        // unwrap ok - another thread took the queue while we were re-acquiring the lock
        // and is guaranteed to process the queue immediately.
        rx.await.unwrap()
    }

    async fn process_queue(&self, buffer: CertBuffer) {
        let committee = self.committee.clone();
        let metrics = self.metrics.clone();
        Handle::current()
            .spawn_blocking(move || Self::process_queue_sync(committee, metrics, buffer))
            .await
            .expect("Spawn blocking should not fail");
    }

    #[instrument(level = "trace", skip_all)]
    fn process_queue_sync(
        committee: Arc<Committee>,
        metrics: Arc<SignatureVerifierMetrics>,
        buffer: CertBuffer,
    ) {
        let _scope = monitored_scope("BatchCertificateVerifier::process_queue");

        let results = batch_verify_certificates(&committee, &buffer.certs.iter().collect_vec());
        izip!(
            results.into_iter(),
            buffer.certs.into_iter(),
            buffer.senders.into_iter(),
        )
        .for_each(|(result, cert, tx)| {
            tx.send(match result {
                Ok(()) => {
                    metrics.total_verified_certs.inc();
                    Ok(VerifiedCertificate::new_unchecked(cert))
                }
                Err(e) => {
                    metrics.total_failed_certs.inc();
                    Err(e)
                }
            })
            .ok();
        });
    }

    #[instrument(level = "trace", skip_all, fields(tx_digest = ?signed_tx.digest()))]
    pub fn verify_tx(&self, signed_tx: &SenderSignedTransaction) -> IotaResult {
        self.signed_data_cache.is_verified(
            signed_tx.full_message_digest(),
            || verify_sender_signed_data_message_signatures(signed_tx, &self.verify_params),
            || Ok(()),
        )
    }

    #[instrument(level = "trace", skip_all)]
    pub fn verify_authority_capabilities(
        &self,
        signed_authority_capabilities: &SignedAuthorityCapabilitiesV1,
    ) -> IotaResult {
        let epoch = self.committee.epoch();
        self.authority_capability_cache.is_verified(
            signed_authority_capabilities.cache_digest(epoch),
            || {
                // Check if authority exists in non-committee validators
                let authority_name = signed_authority_capabilities.data().authority;
                if !self.non_committee_validators.contains(&authority_name) {
                    return Err(IotaError::IncorrectSigner {
                        error: "Signer must be part of non-committee active validators".to_string(),
                    });
                }

                // Create a verification obligation
                let mut obligation = VerificationObligation::default();
                let idx = obligation.add_message(
                    signed_authority_capabilities.data(),
                    epoch, /* epoch is shared between the committee and
                            * non-committee validators */
                    Intent::iota_app(signed_authority_capabilities.scope()),
                );

                // Add the signature and public key to the obligation
                let authority_key = AuthorityPublicKey::from_bytes(authority_name.as_bytes())
                    .map_err(|_| IotaError::IncorrectSigner {
                        error: "Invalid authority public key bytes".to_string(),
                    })?;
                obligation
                    .public_keys
                    .get_mut(idx)
                    .ok_or(IotaError::InvalidAuthenticator)?
                    .push(&authority_key);

                obligation
                    .signatures
                    .get_mut(idx)
                    .ok_or(IotaError::InvalidAuthenticator)?
                    .add_signature(signed_authority_capabilities.auth_sig().clone())
                    .map_err(|_| IotaError::InvalidSignature {
                        error: "Failed to add authority signature to obligation".to_string(),
                    })?;

                obligation.verify_all()
            },
            || Ok(()),
        )
    }

    pub fn clear_signature_cache(&self) {
        self.certificate_cache.clear();
        self.authority_capability_cache.clear();
        self.signed_data_cache.clear();
    }
}

pub struct SignatureVerifierMetrics {
    pub certificate_signatures_cache_hits: IntCounter,
    pub certificate_signatures_cache_misses: IntCounter,
    pub certificate_signatures_cache_evictions: IntCounter,
    pub signed_data_cache_hits: IntCounter,
    pub signed_data_cache_misses: IntCounter,
    pub signed_data_cache_evictions: IntCounter,
    pub authority_capabilities_cache_hits: IntCounter,
    pub authority_capabilities_cache_misses: IntCounter,
    pub authority_capabilities_cache_evictions: IntCounter,
    timeouts: IntCounter,
    full_batches: IntCounter,
    partial_batches: IntCounter,
    total_verified_certs: IntCounter,
    total_failed_certs: IntCounter,
}

impl SignatureVerifierMetrics {
    pub fn new(registry: &Registry) -> Arc<Self> {
        Arc::new(Self {
            certificate_signatures_cache_hits: register_int_counter_with_registry!(
                "certificate_signatures_cache_hits",
                "Number of certificates which were known to be verified because of signature cache.",
                registry
            )
            .unwrap(),
            certificate_signatures_cache_misses: register_int_counter_with_registry!(
                "certificate_signatures_cache_misses",
                "Number of certificates which missed the signature cache",
                registry
            )
            .unwrap(),
            certificate_signatures_cache_evictions: register_int_counter_with_registry!(
                "certificate_signatures_cache_evictions",
                "Number of times we evict a pre-existing key were known to be verified because of signature cache.",
                registry
            )
            .unwrap(),
            signed_data_cache_hits: register_int_counter_with_registry!(
                "signed_data_cache_hits",
                "Number of signed data which were known to be verified because of signature cache.",
                registry
            )
            .unwrap(),
            signed_data_cache_misses: register_int_counter_with_registry!(
                "signed_data_cache_misses",
                "Number of signed data which missed the signature cache.",
                registry
            )
            .unwrap(),
            signed_data_cache_evictions: register_int_counter_with_registry!(
                "signed_data_cache_evictions",
                "Number of times we evict a pre-existing signed data were known to be verified because of signature cache.",
                registry
            )
            .unwrap(),
            authority_capabilities_cache_hits: register_int_counter_with_registry!(
                "authority_capabilities_cache_hits",
                "Number of authority capabilities which were known to be verified because of capabilities cache.",
                registry
            )
            .unwrap(),
            authority_capabilities_cache_misses: register_int_counter_with_registry!(
                "authority_capabilities_cache_misses",
                "Number of authority capabilities which missed the capabilities cache.",
                registry
            )
            .unwrap(),
            authority_capabilities_cache_evictions: register_int_counter_with_registry!(
                "authority_capabilities_cache_evictions",
                "Number of times we evict a pre-existing authority capabilities that were known to be verified.",
                registry
            )
            .unwrap(),
            timeouts: register_int_counter_with_registry!(
                "async_batch_verifier_timeouts",
                "Number of times batch verifier times out and verifies a partial batch",
                registry
            )
            .unwrap(),
            full_batches: register_int_counter_with_registry!(
                "async_batch_verifier_full_batches",
                "Number of times batch verifier verifies a full batch",
                registry
            )
            .unwrap(),
            partial_batches: register_int_counter_with_registry!(
                "async_batch_verifier_partial_batches",
                "Number of times batch verifier verifies a partial batch",
                registry
            )
            .unwrap(),
            total_verified_certs: register_int_counter_with_registry!(
                "async_batch_verifier_total_verified_certs",
                "Total number of certs batch verifier has verified",
                registry
            )
            .unwrap(),
            total_failed_certs: register_int_counter_with_registry!(
                "async_batch_verifier_total_failed_certs",
                "Total number of certs batch verifier has rejected",
                registry
            )
            .unwrap(),
        })
    }
}

/// Verifies all certificates - if any fail return error.
#[instrument(level = "trace", skip_all)]
pub fn batch_verify_all_certificates_and_checkpoints(
    committee: &Committee,
    certs: &[&CertifiedTransaction],
    checkpoints: &[&SignedCheckpointSummary],
) -> IotaResult {
    // certs.data() is assumed to be verified already by the caller.

    for ckpt in checkpoints {
        ckpt.data().verify_epoch(committee.epoch())?;
    }

    batch_verify(committee, certs, checkpoints)
}

/// Verifies certificates in batch mode, but returns a separate result for each
/// cert.
#[instrument(level = "trace", skip_all)]
pub fn batch_verify_certificates(
    committee: &Committee,
    certs: &[&CertifiedTransaction],
) -> Vec<IotaResult> {
    // certs.data() is assumed to be verified already by the caller.
    let verify_params = VerifyParams::default();
    match batch_verify(committee, certs, &[]) {
        Ok(_) => vec![Ok(()); certs.len()],

        // Verify one by one to find which certs were invalid.
        Err(_) if certs.len() > 1 => certs
            .iter()
            // TODO: verify_signature currently checks the tx sig as well, which might be cached
            // already.
            .map(|c| c.verify_signatures_authenticated(committee, &verify_params))
            .collect(),

        Err(e) => vec![Err(e)],
    }
}

fn batch_verify(
    committee: &Committee,
    certs: &[&CertifiedTransaction],
    checkpoints: &[&SignedCheckpointSummary],
) -> IotaResult {
    let mut obligation = VerificationObligation::default();

    for cert in certs {
        let idx = obligation.add_message(cert.data(), cert.epoch(), Intent::iota_app(cert.scope()));
        cert.auth_sig()
            .add_to_verification_obligation(committee, &mut obligation, idx)?;
    }

    for ckpt in checkpoints {
        let idx = obligation.add_message(ckpt.data(), ckpt.epoch(), Intent::iota_app(ckpt.scope()));
        ckpt.auth_sig()
            .add_to_verification_obligation(committee, &mut obligation, idx)?;
    }

    obligation.verify_all()
}