rings_core/message/
payload.rs

1#![warn(missing_docs)]
2
3use std::io::Write;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use bytes::Bytes;
8use derivative::Derivative;
9use flate2::write::GzDecoder;
10use flate2::write::GzEncoder;
11use flate2::Compression;
12use serde::de::DeserializeOwned;
13use serde::Deserialize;
14use serde::Serialize;
15
16use super::encoder::Decoder;
17use super::encoder::Encoded;
18use super::encoder::Encoder;
19use super::protocols::MessageRelay;
20use super::protocols::MessageVerification;
21use super::protocols::MessageVerificationExt;
22use crate::dht::Chord;
23use crate::dht::Did;
24use crate::dht::PeerRing;
25use crate::dht::PeerRingAction;
26use crate::ecc::keccak256;
27use crate::error::Error;
28use crate::error::Result;
29use crate::session::SessionSk;
30
31/// Compresses the given data byte slice using the gzip algorithm with the specified compression level.
32pub fn encode_data_gzip(data: &Bytes, level: u8) -> Result<Bytes> {
33    let mut ec = GzEncoder::new(Vec::new(), Compression::new(level as u32));
34    tracing::info!("data before gzip len: {}", data.len());
35    ec.write_all(data).map_err(|_| Error::GzipEncode)?;
36    ec.finish().map(Bytes::from).map_err(|_| Error::GzipEncode)
37}
38
39/// Serializes the given data using JSON and compresses it with gzip using the specified compression level.
40pub fn gzip_data<T>(data: &T, level: u8) -> Result<Bytes>
41where T: Serialize {
42    let json_bytes = serde_json::to_vec(data).map_err(|_| Error::SerializeToString)?;
43    encode_data_gzip(&json_bytes.into(), level)
44}
45
46/// Decompresses the given gzip-compressed byte slice and returns the decompressed byte slice.
47pub fn decode_gzip_data(data: &Bytes) -> Result<Bytes> {
48    let mut writer = Vec::new();
49    let mut decoder = GzDecoder::new(writer);
50    decoder.write_all(data).map_err(|_| Error::GzipDecode)?;
51    decoder.try_finish().map_err(|_| Error::GzipDecode)?;
52    writer = decoder.finish().map_err(|_| Error::GzipDecode)?;
53    Ok(writer.into())
54}
55
56/// From gzip data to deserialized
57pub fn from_gzipped_data<T>(data: &Bytes) -> Result<T>
58where T: DeserializeOwned {
59    let data = decode_gzip_data(data)?;
60    let m = serde_json::from_slice(&data).map_err(Error::Deserialize)?;
61    Ok(m)
62}
63
64fn hash_transaction(destination: Did, tx_id: uuid::Uuid, data: &[u8]) -> [u8; 32] {
65    let mut msg = vec![];
66
67    msg.extend_from_slice(destination.as_bytes());
68    msg.extend_from_slice(tx_id.as_bytes());
69    msg.extend_from_slice(data);
70
71    keccak256(&msg)
72}
73
74/// All messages transmitted in RingsNetwork should be wrapped by `Transaction`.
75/// It additionally offer destination, tx_id and verification.
76///
77/// To transmit `Transaction` in RingsNetwork, user should build
78/// [MessagePayload] and use [PayloadSender] to send.
79#[derive(Derivative, Deserialize, Serialize, Clone, PartialEq, Eq)]
80#[derivative(Debug)]
81pub struct Transaction {
82    /// The destination of this message.
83    pub destination: Did,
84    /// The transaction ID.
85    /// Remote peer should use same tx_id when response.
86    pub tx_id: uuid::Uuid,
87    /// data
88    pub data: Vec<u8>,
89    /// This field holds a signature from a node,
90    /// which is used to prove that the transaction was created by that node.
91    #[derivative(Debug = "ignore")]
92    pub verification: MessageVerification,
93}
94
95/// `MessagePayload` is used to transmit data between nodes.
96/// The data should be packed by [Transaction].
97#[derive(Derivative, Deserialize, Serialize, Clone, PartialEq, Eq)]
98#[derivative(Debug)]
99pub struct MessagePayload {
100    /// Payload data
101    pub transaction: Transaction,
102    /// Relay records the transport path of message.
103    /// And can also help message sender to find the next hop.
104    pub relay: MessageRelay,
105    /// This field holds a signature from a node,
106    /// which is used to prove that payload was created by that node.
107    #[derivative(Debug = "ignore")]
108    pub verification: MessageVerification,
109}
110
111impl Transaction {
112    /// Wrap data. Will serialize by [bincode::serialize]
113    /// then sign [MessageVerification] by session_sk.
114    pub fn new<T>(
115        destination: Did,
116        tx_id: uuid::Uuid,
117        data: T,
118        session_sk: &SessionSk,
119    ) -> Result<Self>
120    where
121        T: Serialize,
122    {
123        let data = bincode::serialize(&data).map_err(Error::BincodeSerialize)?;
124        let msg_hash = hash_transaction(destination, tx_id, &data);
125        let verification = MessageVerification::new(&msg_hash, session_sk)?;
126        Ok(Self {
127            destination,
128            tx_id,
129            data,
130            verification,
131        })
132    }
133
134    /// Deserializes the data field into a `T` instance.
135    pub fn data<T>(&self) -> Result<T>
136    where T: DeserializeOwned {
137        bincode::deserialize(&self.data).map_err(Error::BincodeDeserialize)
138    }
139}
140
141impl MessagePayload {
142    /// Create new `MessagePayload`.
143    /// Need [Transaction], [SessionSk] and [MessageRelay].
144    pub fn new(
145        transaction: Transaction,
146        session_sk: &SessionSk,
147        relay: MessageRelay,
148    ) -> Result<Self> {
149        let msg_hash = hash_transaction(
150            transaction.destination,
151            transaction.tx_id,
152            &transaction.data,
153        );
154        let verification = MessageVerification::new(&msg_hash, session_sk)?;
155        Ok(Self {
156            transaction,
157            relay,
158            verification,
159        })
160    }
161
162    /// Helps to create sending message from data.
163    pub fn new_send<T>(
164        data: T,
165        session_sk: &SessionSk,
166        next_hop: Did,
167        destination: Did,
168    ) -> Result<Self>
169    where
170        T: Serialize,
171    {
172        let tx_id = uuid::Uuid::new_v4();
173        let transaction = Transaction::new(destination, tx_id, data, session_sk)?;
174        let relay = MessageRelay::new(
175            vec![session_sk.account_did()],
176            next_hop,
177            transaction.destination,
178        );
179        Self::new(transaction, session_sk, relay)
180    }
181
182    /// Deserializes a `MessagePayload` instance from the given binary data.
183    pub fn from_bincode(data: &[u8]) -> Result<Self> {
184        bincode::deserialize(data).map_err(Error::BincodeDeserialize)
185    }
186
187    /// Serializes the `MessagePayload` instance into binary data.
188    pub fn to_bincode(&self) -> Result<Bytes> {
189        bincode::serialize(self)
190            .map(Bytes::from)
191            .map_err(Error::BincodeSerialize)
192    }
193
194    /// Returns whether `local` is the relay destination of this payload.
195    pub(crate) fn is_relay_destination_for(&self, local: Did) -> bool {
196        self.relay.destination == local
197    }
198
199    /// Returns whether `local` should forward this payload to another node.
200    pub(crate) fn should_forward_from(&self, local: Did) -> bool {
201        !self.is_relay_destination_for(local)
202    }
203}
204
205impl MessageVerificationExt for Transaction {
206    fn verification_data(&self) -> Result<Vec<u8>> {
207        Ok(hash_transaction(self.destination, self.tx_id, &self.data).to_vec())
208    }
209
210    fn verification(&self) -> &MessageVerification {
211        &self.verification
212    }
213}
214
215impl MessageVerificationExt for MessagePayload {
216    fn verification_data(&self) -> Result<Vec<u8>> {
217        Ok(hash_transaction(
218            self.transaction.destination,
219            self.transaction.tx_id,
220            &self.transaction.data,
221        )
222        .to_vec())
223    }
224
225    fn verification(&self) -> &MessageVerification {
226        &self.verification
227    }
228}
229
230impl Encoder for MessagePayload {
231    fn encode(&self) -> Result<Encoded> {
232        self.to_bincode()?.encode()
233    }
234}
235
236impl Decoder for MessagePayload {
237    fn from_encoded(encoded: &Encoded) -> Result<Self> {
238        let v: Bytes = encoded.decode()?;
239        Self::from_bincode(&v)
240    }
241}
242
243/// Trait of PayloadSender
244#[cfg_attr(feature = "wasm", async_trait(?Send))]
245#[cfg_attr(not(feature = "wasm"), async_trait)]
246pub trait PayloadSender {
247    /// Get the session sk
248    fn session_sk(&self) -> &SessionSk;
249
250    /// Get access to DHT.
251    fn dht(&self) -> Arc<PeerRing>;
252
253    /// Used to check if destination is already connected when `infer_next_hop`
254    fn is_connected(&self, did: Did) -> bool;
255
256    /// Send a message payload to a specified DID.
257    async fn do_send_payload(&self, did: Did, payload: MessagePayload) -> Result<()>;
258
259    /// Infer the next hop for a message by calling `dht.find_successor()`.
260    fn infer_next_hop(&self, destination: Did, next_hop: Option<Did>) -> Result<Did> {
261        if self.is_connected(destination) {
262            return Ok(destination);
263        }
264
265        if let Some(next_hop) = next_hop {
266            return Ok(next_hop);
267        }
268
269        match self.dht().find_successor(destination)? {
270            PeerRingAction::Some(did) => Ok(did),
271            PeerRingAction::RemoteAction(did, _) => Ok(did),
272            _ => Err(Error::NoNextHop),
273        }
274    }
275
276    /// Alias for `do_send_payload` that sets the next hop to `payload.relay.next_hop`.
277    async fn send_payload(&self, payload: MessagePayload) -> Result<()> {
278        self.do_send_payload(payload.relay.next_hop, payload).await
279    }
280
281    /// Send a message to a specified destination by specified next hop.
282    async fn send_message_by_hop<T>(
283        &self,
284        msg: T,
285        destination: Did,
286        next_hop: Did,
287    ) -> Result<uuid::Uuid>
288    where
289        T: Serialize + Send,
290    {
291        let payload = MessagePayload::new_send(msg, self.session_sk(), next_hop, destination)?;
292        let tx_id = payload.transaction.tx_id;
293        self.send_payload(payload).await?;
294        Ok(tx_id)
295    }
296
297    /// Send a message to a specified destination.
298    async fn send_message<T>(&self, msg: T, destination: Did) -> Result<uuid::Uuid>
299    where T: Serialize + Send {
300        let next_hop = self.infer_next_hop(destination, None)?;
301        self.send_message_by_hop(msg, destination, next_hop).await
302    }
303
304    /// Send a direct message to a specified destination.
305    async fn send_direct_message<T>(&self, msg: T, destination: Did) -> Result<uuid::Uuid>
306    where T: Serialize + Send {
307        self.send_message_by_hop(msg, destination, destination)
308            .await
309    }
310
311    /// Send a report message to a specified destination.
312    async fn send_report_message<T>(&self, payload: &MessagePayload, msg: T) -> Result<()>
313    where T: Serialize + Send {
314        let relay = payload.relay.report(self.dht().did)?;
315
316        let transaction = Transaction::new(
317            relay.destination,
318            payload.transaction.tx_id,
319            msg,
320            self.session_sk(),
321        )?;
322
323        let pl = MessagePayload::new(transaction, self.session_sk(), relay)?;
324        self.send_payload(pl).await
325    }
326
327    /// Forward a payload message by relay.
328    /// It just create a new payload, cloned data, resigned with session and send
329    async fn forward_by_relay(&self, payload: &MessagePayload, relay: MessageRelay) -> Result<()> {
330        let new_pl = MessagePayload::new(payload.transaction.clone(), self.session_sk(), relay)?;
331        self.send_payload(new_pl).await
332    }
333
334    /// Forward a payload message, with the next hop inferred by the DHT.
335    async fn forward_payload(&self, payload: &MessagePayload, next_hop: Option<Did>) -> Result<()> {
336        let next_hop = self.infer_next_hop(payload.relay.destination, next_hop)?;
337        let relay = payload.relay.forward(self.dht().did, next_hop)?;
338        self.forward_by_relay(payload, relay).await
339    }
340
341    /// Reset the destination to a secp DID.
342    async fn reset_destination(&self, payload: &MessagePayload, next_hop: Did) -> Result<()> {
343        let relay = payload
344            .relay
345            .reset_destination(next_hop)
346            .forward(self.dht().did, next_hop)?;
347        self.forward_by_relay(payload, relay).await
348    }
349}
350
351#[cfg(test)]
352pub mod test {
353    use rand::Rng;
354
355    use super::*;
356    use crate::ecc::SecretKey;
357    use crate::message::Message;
358
359    #[derive(Deserialize, Serialize, PartialEq, Debug, Clone)]
360    pub struct TestData {
361        a: String,
362        b: i64,
363        c: f64,
364        d: bool,
365    }
366
367    pub fn new_test_payload(next_hop: Did) -> MessagePayload {
368        let test_data = TestData {
369            a: "hello".to_string(),
370            b: 111,
371            c: 2.33,
372            d: true,
373        };
374        new_payload(test_data, next_hop)
375    }
376
377    pub fn new_payload<T>(data: T, next_hop: Did) -> MessagePayload
378    where T: Serialize + DeserializeOwned {
379        let key = SecretKey::random();
380        let destination = SecretKey::random().address().into();
381        let session_sk = SessionSk::new_with_seckey(&key).unwrap();
382        MessagePayload::new_send(data, &session_sk, next_hop, destination).unwrap()
383    }
384
385    #[test]
386    fn new_then_verify() {
387        let key2 = SecretKey::random();
388        let did2 = key2.address().into();
389
390        let payload = new_test_payload(did2);
391        assert!(payload.verify());
392    }
393
394    #[test]
395    fn relay_destination_predicates_name_forwarding_state() -> Result<()> {
396        let local_key = SecretKey::random();
397        let session_sk = SessionSk::new_with_seckey(&local_key)?;
398        let local: Did = local_key.address().into();
399        let remote: Did = SecretKey::random().address().into();
400
401        let local_payload =
402            MessagePayload::new_send(Message::custom(b"local")?, &session_sk, local, local)?;
403        assert!(local_payload.is_relay_destination_for(local));
404        assert!(!local_payload.should_forward_from(local));
405
406        let remote_payload =
407            MessagePayload::new_send(Message::custom(b"remote")?, &session_sk, remote, remote)?;
408        assert!(!remote_payload.is_relay_destination_for(local));
409        assert!(remote_payload.should_forward_from(local));
410
411        Ok(())
412    }
413
414    /// The sender cuts chunk data at `max_message_size - (MAX_CHUNK_ENVELOPE_OVERHEAD +
415    /// TRANSPORT_CUSTOM_OVERHEAD)`. This pins that those reserves are large enough by measuring the
416    /// *exact* bytes the data channel carries: a full-size chunk, re-wrapped in its `MessagePayload`
417    /// **and** the outer `TransportMessage::Custom` frame (what `send_data` actually serializes),
418    /// stays at or below `MAX_DATA_CHANNEL_MESSAGE_SIZE`. If either envelope grows past its reserve,
419    /// this fails instead of silently producing oversized frames the channel would reject.
420    #[test]
421    fn chunk_envelope_fits_reserve() {
422        use rings_transport::core::transport::TransportMessage;
423        use rings_transport::core::transport::MAX_DATA_CHANNEL_MESSAGE_SIZE;
424
425        use crate::chunk::ChunkList;
426        use crate::consts::MAX_CHUNK_ENVELOPE_OVERHEAD;
427        use crate::consts::TRANSPORT_CUSTOM_OVERHEAD;
428
429        let next_hop = SecretKey::random().address().into();
430        let chunk_size = MAX_DATA_CHANNEL_MESSAGE_SIZE
431            - (MAX_CHUNK_ENVELOPE_OVERHEAD + TRANSPORT_CUSTOM_OVERHEAD);
432        let data: Bytes = vec![0xab; chunk_size].into();
433        let chunk = ChunkList::split(&data, chunk_size)
434            .to_vec()
435            .pop()
436            .expect("one chunk");
437
438        // The bytes actually handed to SCTP: bincode(Custom(bincode(MessagePayload))).
439        let payload_bytes = new_payload(Message::Chunk(chunk), next_hop)
440            .to_bincode()
441            .unwrap();
442        let wire = bincode::serialize(&TransportMessage::Custom(payload_bytes.to_vec())).unwrap();
443
444        assert!(
445            wire.len() <= MAX_DATA_CHANNEL_MESSAGE_SIZE,
446            "wrapped chunk frame is {} bytes, exceeds limit {}; raise the reserves",
447            wire.len(),
448            MAX_DATA_CHANNEL_MESSAGE_SIZE,
449        );
450    }
451
452    /// The other framing boundary: a payload [`WireReserves::plan`] keeps `Whole`, once wrapped in
453    /// the outer `TransportMessage::Custom` frame, stays within the limit — pinning that
454    /// `WireReserves::PRODUCTION.whole` is enough for the whole-message path (not just the chunk
455    /// path), and that one byte past the boundary switches to chunked.
456    #[test]
457    fn whole_message_boundary_fits_custom_wrapper() {
458        use rings_transport::core::transport::TransportMessage;
459        use rings_transport::core::transport::MAX_DATA_CHANNEL_MESSAGE_SIZE;
460
461        use crate::chunk::Framing;
462        use crate::chunk::WireReserves;
463
464        let reserves = WireReserves::PRODUCTION;
465        let limit = MAX_DATA_CHANNEL_MESSAGE_SIZE;
466        // Largest payload that should still be sent whole.
467        let payload_len = limit - reserves.whole;
468        assert_eq!(reserves.plan(payload_len, limit), Some(Framing::Whole));
469
470        let wire = bincode::serialize(&TransportMessage::Custom(vec![0u8; payload_len])).unwrap();
471        assert!(
472            wire.len() <= limit,
473            "whole wire {} exceeds limit {}",
474            wire.len(),
475            limit
476        );
477        // One byte past the boundary must switch to chunked.
478        assert!(matches!(
479            reserves.plan(payload_len + 1, limit),
480            Some(Framing::Chunked { .. })
481        ));
482    }
483
484    #[test]
485    fn test_message_payload_from_auto() {
486        let next_hop = SecretKey::random().address().into();
487
488        let payload = new_test_payload(next_hop);
489        let gzipped_encoded_payload = payload.encode().unwrap();
490        let payload2: MessagePayload = gzipped_encoded_payload.decode().unwrap();
491        assert_eq!(payload, payload2);
492
493        let gunzip_encoded_payload = payload.to_bincode().unwrap().encode().unwrap();
494        let payload2: MessagePayload = gunzip_encoded_payload.decode().unwrap();
495        assert_eq!(payload, payload2);
496    }
497
498    #[test]
499    fn test_message_payload_encode_len() {
500        let next_hop = SecretKey::random().address().into();
501        let data = rand::thread_rng().gen::<[u8; 32]>();
502
503        let data1 = data;
504        let msg1 = Message::custom(&data1).unwrap();
505        let payload1 = new_payload(msg1, next_hop);
506        let bytes1 = payload1.to_bincode().unwrap();
507        let encoded1 = payload1.encode().unwrap();
508        let encoded_bytes1: Vec<u8> = encoded1.into();
509
510        let data2 = data.repeat(2);
511        let msg2 = Message::custom(&data2).unwrap();
512        let payload2 = new_payload(msg2, next_hop);
513        let bytes2 = payload2.to_bincode().unwrap();
514        let encoded2 = payload2.encode().unwrap();
515        let encoded_bytes2: Vec<u8> = encoded2.into();
516
517        assert_eq!(bytes1.len() - data1.len(), bytes2.len() - data2.len());
518        assert_ne!(
519            encoded_bytes1.len() - data1.len(),
520            encoded_bytes2.len() - data2.len()
521        );
522    }
523}
rings_core/message/payload.rs

rings_core/message/
payload.rs
