massa_bootstrap/bindings/
client.rs

1// Copyright (c) 2022 MASSA LABS <info@massa.net>
2
3use crate::bindings::{BindingReadExact, BindingWriteExact};
4use crate::error::BootstrapError;
5use crate::messages::{
6    BootstrapClientMessage, BootstrapClientMessageSerializer, BootstrapServerMessage,
7    BootstrapServerMessageDeserializer,
8};
9use crate::settings::BootstrapClientConfig;
10use massa_hash::Hash;
11use massa_models::config::{
12    BOOTSTRAP_MESSAGE_LEN_PREFIX_MAX, BOOTSTRAP_MESSAGE_LEN_PREFIX_SIZE_BYTES,
13    MAX_BOOTSTRAP_MESSAGE_FROM_CLIENT_SIZE, MAX_BOOTSTRAP_MESSAGE_FROM_SERVER_SIZE,
14    SIGNATURE_DESER_SIZE,
15};
16use massa_models::serialization::{DeserializeMinBEInt, SerializeMinBEInt};
17use massa_models::version::{Version, VersionSerializer};
18use massa_serialization::{DeserializeError, Deserializer, Serializer};
19use massa_signature::{PublicKey, Signature};
20use rand::{rngs::StdRng, RngCore, SeedableRng};
21use std::time::Instant;
22use std::{net::TcpStream, time::Duration};
23use stream_limiter::{Limiter, LimiterOptions};
24
25/// Bootstrap client binder
26pub struct BootstrapClientBinder {
27    remote_pubkey: PublicKey,
28    duplex: Limiter<TcpStream>,
29    prev_message: Option<Hash>,
30    version_serializer: VersionSerializer,
31    cfg: BootstrapClientConfig,
32    /// Cached from the first bootstrap part; consensus blocks only stream after final state.
33    last_start_period: Option<u64>,
34}
35
36const KNOWN_PREFIX_FROM_SERVER_LEN: usize =
37    SIGNATURE_DESER_SIZE + BOOTSTRAP_MESSAGE_LEN_PREFIX_SIZE_BYTES;
38/// The known-length component of a message to be received.
39struct ServerMessageLeader {
40    sig: Signature,
41    msg_len: u32,
42}
43
44impl BootstrapClientBinder {
45    /// Creates a new `WriteBinder`.
46    ///
47    /// # Argument
48    /// * duplex: duplex stream.
49    /// * limit: limit max bytes per second (up and down)
50    #[allow(clippy::too_many_arguments)]
51    pub fn new(
52        duplex: TcpStream,
53        remote_pubkey: PublicKey,
54        cfg: BootstrapClientConfig,
55        limit: Option<u64>,
56    ) -> Self {
57        let limit_opts =
58            limit.map(|limit| LimiterOptions::new(limit, Duration::from_millis(1000), limit));
59        let duplex = Limiter::new(duplex, limit_opts.clone(), limit_opts);
60        BootstrapClientBinder {
61            remote_pubkey,
62            duplex,
63            prev_message: None,
64            version_serializer: VersionSerializer::new(),
65            cfg,
66            last_start_period: None,
67        }
68    }
69
70    /// Update the last start period used when deserializing bootstrap block headers.
71    pub fn set_last_start_period(&mut self, last_start_period: Option<u64>) {
72        self.last_start_period = last_start_period;
73    }
74
75    /// Performs a handshake. Should be called after connection
76    /// NOT cancel-safe
77    pub fn handshake(&mut self, version: Version) -> Result<(), BootstrapError> {
78        // send version and randomn bytes
79        let msg_hash = {
80            let mut version_ser = Vec::new();
81            self.version_serializer
82                .serialize(&version, &mut version_ser)?;
83            let mut version_random_bytes =
84                vec![0u8; version_ser.len() + self.cfg.randomness_size_bytes];
85            version_random_bytes[..version_ser.len()].clone_from_slice(&version_ser);
86            StdRng::from_entropy().fill_bytes(&mut version_random_bytes[version_ser.len()..]);
87            self.write_all_timeout(&version_random_bytes, None)
88                .map_err(|(e, _)| e)?;
89            Hash::compute_from(&version_random_bytes)
90        };
91
92        self.prev_message = Some(msg_hash);
93
94        Ok(())
95    }
96
97    /// Reads the next message.
98    pub fn next_timeout(
99        &mut self,
100        duration: Option<Duration>,
101    ) -> Result<BootstrapServerMessage, BootstrapError> {
102        let deadline = duration.map(|d| Instant::now() + d);
103
104        // read the known-len component of the message
105        let mut known_len_buff = [0u8; KNOWN_PREFIX_FROM_SERVER_LEN];
106        // TODO: handle a partial read
107        self.read_exact_timeout(&mut known_len_buff, deadline)
108            .map_err(|(err, _consumed)| err)?;
109
110        let ServerMessageLeader { sig, msg_len } = self.decode_msg_leader(&known_len_buff)?;
111
112        // Update this bindings "most recently received" message hash, retaining the replaced value
113        let message_deserializer = BootstrapServerMessageDeserializer::with_last_start_period(
114            (&self.cfg).into(),
115            self.last_start_period,
116        );
117        let prev_msg = self
118            .prev_message
119            .replace(Hash::compute_from(&sig.to_bytes()));
120
121        let message = {
122            if let Some(prev_msg) = prev_msg {
123                // Consume the rest of the message from the stream
124                let mut stream_bytes =
125                    vec![0u8; msg_len.try_into().expect("Overflow on msg_len to usize")];
126
127                // TODO: handle a partial read
128                self.read_exact_timeout(&mut stream_bytes[..], deadline)
129                    .map_err(|(e, _consumed)| e)?;
130                let msg_bytes = &mut stream_bytes[..];
131
132                // prepend the received message with the previous messages hash, and derive the new hash.
133                // TODO: some sort of recovery if this fails?
134                let rehash_seed = &[prev_msg.to_bytes().as_slice(), msg_bytes].concat();
135                let msg_hash = Hash::compute_from(rehash_seed);
136                self.remote_pubkey.verify_signature(&msg_hash, &sig)?;
137
138                // ...And deserialize
139                let (_, msg) = message_deserializer
140                    .deserialize::<DeserializeError>(msg_bytes)
141                    .map_err(|err| BootstrapError::DeserializeError(format!("{}", err)))?;
142                msg
143            } else {
144                // Consume the rest of the message from the stream
145                let mut stream_bytes =
146                    vec![0u8; msg_len.try_into().expect("Overflow on msg_len to usize")];
147
148                // TODO: handle a partial read
149                self.read_exact_timeout(&mut stream_bytes[..], deadline)
150                    .map_err(|(e, _)| e)?;
151                let sig_msg_bytes = &mut stream_bytes[..];
152
153                // Compute the hash and verify
154                let msg_hash = Hash::compute_from(sig_msg_bytes);
155                self.remote_pubkey.verify_signature(&msg_hash, &sig)?;
156
157                // ...And deserialize
158                let (_, msg) = message_deserializer
159                    .deserialize::<DeserializeError>(sig_msg_bytes)
160                    .map_err(|err| BootstrapError::DeserializeError(format!("{}", err)))?;
161                msg
162            }
163        };
164        Ok(message)
165    }
166
167    // TODO: use a proper (de)serializer: https://github.com/massalabs/massa/pull/3745#discussion_r1169733161
168    /// Send a message to the bootstrap server
169    pub fn send_timeout(
170        &mut self,
171        msg: &BootstrapClientMessage,
172        duration: Option<Duration>,
173    ) -> Result<(), BootstrapError> {
174        let deadline = duration.map(|d| Instant::now() + d);
175        let mut msg_bytes = Vec::new();
176        let message_serializer = BootstrapClientMessageSerializer::new();
177        message_serializer.serialize(msg, &mut msg_bytes)?;
178        let msg_len: u32 = msg_bytes.len().try_into().map_err(|e| {
179            BootstrapError::GeneralError(format!("bootstrap message too large to encode: {}", e))
180        })?;
181
182        let mut write_buf = Vec::new();
183        if let Some(prev_message) = self.prev_message {
184            // there was a previous message
185            let prev_message = prev_message.to_bytes();
186
187            // update current previous message to be hash(prev_msg_hash + msg)
188            let mut hash_data =
189                Vec::with_capacity(prev_message.len().saturating_add(msg_bytes.len()));
190            hash_data.extend(prev_message);
191            hash_data.extend(&msg_bytes);
192            self.prev_message = Some(Hash::compute_from(&hash_data));
193
194            // Provide the signature saved as the previous message
195            write_buf.extend(prev_message);
196        } else {
197            // No previous message, so we set the hash-chain genesis to the hash of the first msg
198            self.prev_message = Some(Hash::compute_from(&msg_bytes));
199        }
200
201        // Enforce the client-side cap on the value, but encode the prefix against
202        // BOOTSTRAP_MESSAGE_LEN_PREFIX_MAX: the width is part of the wire format and the server
203        // reads exactly BOOTSTRAP_MESSAGE_LEN_PREFIX_SIZE_BYTES after the hash
204        // (see BootstrapServerBinder::next_timeout).
205        if msg_len > MAX_BOOTSTRAP_MESSAGE_FROM_CLIENT_SIZE {
206            return Err(BootstrapError::GeneralError(format!(
207                "bootstrap message too large to send: {} > {}",
208                msg_len, MAX_BOOTSTRAP_MESSAGE_FROM_CLIENT_SIZE
209            )));
210        }
211        write_buf.extend(msg_len.to_be_bytes_min(BOOTSTRAP_MESSAGE_LEN_PREFIX_MAX)?);
212
213        // Provide the message
214        write_buf.extend(&msg_bytes);
215
216        // And send it off
217        self.write_all_timeout(&write_buf, deadline)
218            .map_err(|(e, _)| e)?;
219        Ok(())
220    }
221
222    /// We are using this instead of of our library deserializer as the process is relatively straight forward
223    /// and makes error-type management cleaner
224    fn decode_msg_leader(
225        &self,
226        leader_buff: &[u8; KNOWN_PREFIX_FROM_SERVER_LEN],
227    ) -> Result<ServerMessageLeader, BootstrapError> {
228        let sig = Signature::from_bytes(leader_buff)?;
229
230        // construct the message len from the leader-bufff
231        let msg_len = u32::from_be_bytes_min(
232            &leader_buff[SIGNATURE_DESER_SIZE..],
233            MAX_BOOTSTRAP_MESSAGE_FROM_SERVER_SIZE,
234        )?
235        .0;
236        Ok(ServerMessageLeader { sig, msg_len })
237    }
238}
239
240impl crate::bindings::BindingReadExact for BootstrapClientBinder {
241    fn set_read_timeout(&mut self, duration: Option<Duration>) -> Result<(), std::io::Error> {
242        if let Some(ref mut opts) = self.duplex.read_opt {
243            opts.timeout = duration;
244        }
245        self.duplex.stream.set_read_timeout(duration)
246    }
247}
248
249impl std::io::Read for BootstrapClientBinder {
250    fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
251        self.duplex.read(buf)
252    }
253}
254
255impl crate::bindings::BindingWriteExact for BootstrapClientBinder {
256    fn set_write_timeout(&mut self, duration: Option<Duration>) -> Result<(), std::io::Error> {
257        if let Some(ref mut opts) = self.duplex.write_opt {
258            opts.timeout = duration;
259        }
260        self.duplex.stream.set_write_timeout(duration)
261    }
262}
263
264impl std::io::Write for BootstrapClientBinder {
265    fn write(&mut self, buf: &[u8]) -> Result<usize, std::io::Error> {
266        self.duplex.write(buf)
267    }
268
269    fn flush(&mut self) -> Result<(), std::io::Error> {
270        self.duplex.flush()
271    }
272}
massa_bootstrap/bindings/client.rs

massa_bootstrap/bindings/
client.rs
