1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

// Copyright (c) 2022, Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0
use blake2::digest::Update;
use config::{Committee, SharedWorkerCache};
use crypto::PublicKey;
use fastcrypto::{Digest, Hash};
use std::{
    collections::BTreeMap,
    fmt::{Debug, Display, Formatter},
};
use thiserror::Error;
use tracing::{error, warn};
use types::{Certificate, CertificateDigest, Round};

// RequestID helps us identify an incoming request and
// all the consequent network requests associated with it.
#[derive(Clone, Copy, Default, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct RequestID(pub [u8; fastcrypto::DIGEST_LEN]);

impl RequestID {
    // Create a request key (deterministically) from arbitrary data.
    pub fn new(data: &[u8]) -> Self {
        RequestID(fastcrypto::blake2b_256(|hasher| hasher.update(data)))
    }
}

impl Debug for RequestID {
    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> {
        write!(f, "{}", base64::encode(self.0))
    }
}

impl FromIterator<CertificateDigest> for RequestID {
    fn from_iter<T: IntoIterator<Item = CertificateDigest>>(ids: T) -> Self {
        let mut ids_sorted: Vec<CertificateDigest> = ids.into_iter().collect();
        ids_sorted.sort();

        let result: Vec<u8> = ids_sorted
            .into_iter()
            .flat_map(|d| Digest::from(d).to_vec())
            .collect();

        RequestID::new(&result)
    }
}

impl<'a> FromIterator<&'a Certificate> for RequestID {
    fn from_iter<T: IntoIterator<Item = &'a Certificate>>(certificates: T) -> Self {
        certificates.into_iter().map(|c| c.digest()).collect()
    }
}

impl Display for RequestID {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", base64::encode(self.0))
    }
}

#[derive(Debug, Clone)]
pub struct PayloadAvailabilityResponse {
    pub block_ids: Vec<(CertificateDigest, bool)>,
    pub from: PublicKey,
}

impl PayloadAvailabilityResponse {
    pub fn request_id(&self) -> RequestID {
        self.block_ids.iter().map(|entry| entry.0).collect()
    }

    pub fn available_block_ids(&self) -> Vec<CertificateDigest> {
        self.block_ids
            .iter()
            .filter_map(|(id, available)| available.then_some(*id))
            .collect()
    }
}

#[derive(Debug, Clone)]
pub struct CertificatesResponse {
    pub certificates: Vec<(CertificateDigest, Option<Certificate>)>,
    pub from: PublicKey,
}

impl CertificatesResponse {
    pub fn request_id(&self) -> RequestID {
        self.certificates.iter().map(|entry| entry.0).collect()
    }

    /// This method does two things:
    /// 1) filters only the found certificates
    /// 2) validates the certificates
    /// Even if one found certificate is not valid, an error is returned. Otherwise
    /// and Ok result is returned with (any) found certificates.
    pub fn validate_certificates(
        &self,
        committee: &Committee,
        worker_cache: SharedWorkerCache,
    ) -> Result<Vec<Certificate>, CertificatesResponseError> {
        let peer_found_certs: Vec<Certificate> = self
            .certificates
            .iter()
            .filter_map(|e| e.1.clone())
            .collect();

        if peer_found_certs.as_slice().is_empty() {
            // no certificates found, skip
            warn!(
                "No certificates are able to be served from {:?}",
                &self.from
            );
            return Ok(vec![]);
        }

        let invalid_certificates: Vec<Certificate> = peer_found_certs
            .clone()
            .into_iter()
            .filter(|c| {
                if let Err(err) = c.verify(committee, worker_cache.clone()) {
                    error!(
                        "Certificate verification failed for id {} with error {:?}",
                        c.digest(),
                        err
                    );
                    return true;
                }
                false
            })
            .collect();

        if !invalid_certificates.is_empty() {
            error!("Found at least one invalid certificate from peer {:?}. Will ignore all certificates", self.from);

            return Err(CertificatesResponseError::ValidationError {
                name: self.from.clone(),
                invalid_certificates,
            });
        }

        Ok(peer_found_certs)
    }
}

#[derive(Debug, Clone)]
pub struct CertificateDigestsResponse {
    // Certificate digests, grouped by round numbers.
    pub certificate_ids: BTreeMap<Round, Vec<CertificateDigest>>,
    pub from: PublicKey,
}

#[derive(Debug, Clone)]
pub enum AvailabilityResponse {
    CertificateDigest(CertificateDigestsResponse),
    Certificate(CertificatesResponse),
    Payload(PayloadAvailabilityResponse),
}

#[derive(Debug, Error, Clone, PartialEq)]
pub enum CertificatesResponseError {
    #[error("Found invalid certificates form peer {name} - potentially Byzantine.")]
    ValidationError {
        name: PublicKey,
        invalid_certificates: Vec<Certificate>,
    },
}