rings_node/
processor.rs

1#![warn(missing_docs)]
2
3//! Processor of rings-node rpc server.
4
5use std::str::FromStr;
6use std::sync::Arc;
7use std::time::Duration;
8
9use rings_core::chunk::ReassemblyLimits;
10use rings_core::dht::Did;
11use rings_core::dht::EntryStorage;
12use rings_core::dht::DEFAULT_FINGER_TABLE_SIZE;
13use rings_core::ecc::PublicKey;
14use rings_core::ecc::SecretKey;
15use rings_core::measure::MeasureImpl;
16use rings_core::message::e2e;
17use rings_core::message::e2e::E2eHandshakeRequest;
18use rings_core::message::e2e::E2eHandshakeResponse;
19use rings_core::message::e2e::E2eStreamDecryptor;
20use rings_core::message::e2e::E2eStreamFrame;
21use rings_core::message::Encoded;
22use rings_core::message::Encoder;
23use rings_core::message::Message;
24use rings_core::prelude::uuid;
25use rings_core::storage::MemStorage;
26use rings_core::swarm::Swarm;
27use rings_core::swarm::SwarmBuilder;
28use rings_rpc::protos::rings_node::*;
29use rings_transport::webrtc_config::WebrtcUdpPortRange;
30use serde::Deserialize;
31use serde::Serialize;
32
33use crate::consts::DATA_REDUNDANT;
34use crate::error::Error;
35use crate::error::Result;
36use crate::measure::PeriodicMeasure;
37use crate::prelude::entry;
38use crate::prelude::wasm_export;
39use crate::prelude::ChordStorageInterface;
40use crate::prelude::ChordStorageInterfaceCacheChecker;
41use crate::prelude::SessionSk;
42
43/// ProcessorConfig is usually serialized as json or yaml.
44/// There is a `from_config` method in [ProcessorBuilder] used to initialize the Builder with a serialized ProcessorConfig.
45#[derive(Clone, Debug)]
46#[wasm_export]
47pub struct ProcessorConfig {
48    /// The network_id is used to distinguish different networks.
49    /// Use 1 for main network.
50    network_id: u32,
51    /// ICE servers for webrtc
52    ice_servers: String,
53    /// External address for webrtc
54    external_address: Option<String>,
55    /// Inclusive lower native WebRTC UDP port bound.
56    webrtc_udp_port_min: Option<u16>,
57    /// Inclusive upper native WebRTC UDP port bound.
58    webrtc_udp_port_max: Option<u16>,
59    /// [SessionSk].
60    session_sk: SessionSk,
61    /// Stabilization interval.
62    stabilize_interval: Duration,
63}
64
65#[wasm_export]
66impl ProcessorConfig {
67    /// Creates a new `ProcessorConfig` instance without an external address.
68    pub fn new(
69        network_id: u32,
70        ice_servers: String,
71        session_sk: SessionSk,
72        stabilize_interval: u64,
73    ) -> Self {
74        Self {
75            network_id,
76            ice_servers,
77            external_address: None,
78            webrtc_udp_port_min: None,
79            webrtc_udp_port_max: None,
80            session_sk,
81            stabilize_interval: Duration::from_secs(stabilize_interval),
82        }
83    }
84
85    /// Return associated [SessionSk].
86    pub fn session_sk(&self) -> SessionSk {
87        self.session_sk.clone()
88    }
89}
90
91impl ProcessorConfig {
92    /// Returns the validated native WebRTC UDP port range, when configured.
93    pub fn webrtc_udp_port_range(&self) -> Result<Option<WebrtcUdpPortRange>> {
94        parse_webrtc_udp_port_range(self.webrtc_udp_port_min, self.webrtc_udp_port_max)
95    }
96}
97
98impl FromStr for ProcessorConfig {
99    type Err = Error;
100    /// Reveal config from serialized string.
101    fn from_str(ser: &str) -> Result<Self> {
102        serde_yaml::from_str::<ProcessorConfig>(ser).map_err(Error::SerdeYamlError)
103    }
104}
105
106/// `ProcessorConfigSerialized` is a serialized version of `ProcessorConfig`.
107/// Instead of storing the `SessionSk` instance, it stores the dumped string representation of the session secret key.
108#[derive(Serialize, Deserialize, Clone)]
109#[wasm_export]
110pub struct ProcessorConfigSerialized {
111    /// The network_id is used to distinguish different networks.
112    /// Use 1 for main network.
113    network_id: u32,
114    /// A string representing ICE servers for WebRTC
115    ice_servers: String,
116    /// An optional string representing the external address for WebRTC
117    external_address: Option<String>,
118    /// Inclusive lower native WebRTC UDP port bound.
119    webrtc_udp_port_min: Option<u16>,
120    /// Inclusive upper native WebRTC UDP port bound.
121    webrtc_udp_port_max: Option<u16>,
122    /// A string representing the dumped `SessionSk`.
123    session_sk: String,
124    /// An unsigned integer representing the stabilization interval in seconds.
125    stabilize_interval: u64,
126}
127
128impl ProcessorConfigSerialized {
129    /// Creates a new `ProcessorConfigSerialized` instance without an external address.
130    pub fn new(
131        network_id: u32,
132        ice_servers: String,
133        session_sk: String,
134        stabilize_interval: u64,
135    ) -> Self {
136        Self {
137            network_id,
138            ice_servers,
139            external_address: None,
140            webrtc_udp_port_min: None,
141            webrtc_udp_port_max: None,
142            session_sk,
143            stabilize_interval,
144        }
145    }
146
147    /// Sets up the external address for WebRTC.
148    /// This will be used to configure the transport to listen for WebRTC connections in "HOST" mode.
149    pub fn external_address(mut self, external_address: String) -> Self {
150        self.external_address = Some(external_address);
151        self
152    }
153
154    /// Sets the native WebRTC UDP port range bounds.
155    pub fn webrtc_udp_port_range(mut self, range: WebrtcUdpPortRange) -> Self {
156        self.webrtc_udp_port_min = Some(range.min());
157        self.webrtc_udp_port_max = Some(range.max());
158        self
159    }
160}
161
162pub(crate) fn parse_webrtc_udp_port_range(
163    min: Option<u16>,
164    max: Option<u16>,
165) -> Result<Option<WebrtcUdpPortRange>> {
166    match (min, max) {
167        (None, None) => Ok(None),
168        (Some(min), Some(max)) => WebrtcUdpPortRange::new(min, max)
169            .map(Some)
170            .map_err(Error::from),
171        (min, max) => Err(Error::IncompleteWebrtcUdpPortRange { min, max }),
172    }
173}
174
175impl TryFrom<ProcessorConfig> for ProcessorConfigSerialized {
176    type Error = Error;
177    fn try_from(ins: ProcessorConfig) -> Result<Self> {
178        Ok(Self {
179            network_id: ins.network_id,
180            ice_servers: ins.ice_servers.clone(),
181            external_address: ins.external_address.clone(),
182            webrtc_udp_port_min: ins.webrtc_udp_port_min,
183            webrtc_udp_port_max: ins.webrtc_udp_port_max,
184            session_sk: ins.session_sk.dump()?,
185            stabilize_interval: ins.stabilize_interval.as_secs(),
186        })
187    }
188}
189
190impl TryFrom<ProcessorConfigSerialized> for ProcessorConfig {
191    type Error = Error;
192    fn try_from(ins: ProcessorConfigSerialized) -> Result<Self> {
193        let webrtc_udp_port_range =
194            parse_webrtc_udp_port_range(ins.webrtc_udp_port_min, ins.webrtc_udp_port_max)?;
195        Ok(Self {
196            network_id: ins.network_id,
197            ice_servers: ins.ice_servers.clone(),
198            external_address: ins.external_address.clone(),
199            webrtc_udp_port_min: webrtc_udp_port_range.map(WebrtcUdpPortRange::min),
200            webrtc_udp_port_max: webrtc_udp_port_range.map(WebrtcUdpPortRange::max),
201            session_sk: SessionSk::from_str(&ins.session_sk)?,
202            stabilize_interval: Duration::from_secs(ins.stabilize_interval),
203        })
204    }
205}
206
207impl Serialize for ProcessorConfig {
208    fn serialize<S: serde::Serializer>(
209        &self,
210        serializer: S,
211    ) -> core::result::Result<S::Ok, S::Error> {
212        let ins: ProcessorConfigSerialized = self
213            .clone()
214            .try_into()
215            .map_err(|e: Error| serde::ser::Error::custom(e.to_string()))?;
216        ProcessorConfigSerialized::serialize(&ins, serializer)
217    }
218}
219
220impl<'de> serde::de::Deserialize<'de> for ProcessorConfig {
221    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
222    where D: serde::Deserializer<'de> {
223        match ProcessorConfigSerialized::deserialize(deserializer) {
224            Ok(ins) => {
225                let cfg: ProcessorConfig = ins
226                    .try_into()
227                    .map_err(|e: Error| serde::de::Error::custom(e.to_string()))?;
228                Ok(cfg)
229            }
230            Err(e) => Err(e),
231        }
232    }
233}
234
235/// ProcessorBuilder is used to initialize a [Processor] instance.
236pub struct ProcessorBuilder {
237    network_id: u32,
238    ice_servers: String,
239    external_address: Option<String>,
240    webrtc_udp_port_range: Option<WebrtcUdpPortRange>,
241    session_sk: SessionSk,
242    storage: Option<EntryStorage>,
243    measure: Option<MeasureImpl>,
244    stabilize_interval: Duration,
245    dht_finger_table_size: usize,
246    reassembly_limits: ReassemblyLimits,
247}
248
249/// Processor for rings-node rpc server
250#[derive(Clone)]
251pub struct Processor {
252    /// a swarm instance
253    pub swarm: Arc<Swarm>,
254    stabilize_interval: Duration,
255}
256
257impl ProcessorBuilder {
258    /// initialize a [ProcessorBuilder] with a serialized [ProcessorConfig].
259    pub fn from_serialized(config: &str) -> Result<Self> {
260        let config =
261            serde_yaml::from_str::<ProcessorConfig>(config).map_err(Error::SerdeYamlError)?;
262        Self::from_config(&config)
263    }
264
265    /// initialize a [ProcessorBuilder] with a [ProcessorConfig].
266    pub fn from_config(config: &ProcessorConfig) -> Result<Self> {
267        Ok(Self {
268            network_id: config.network_id,
269            ice_servers: config.ice_servers.clone(),
270            external_address: config.external_address.clone(),
271            webrtc_udp_port_range: config.webrtc_udp_port_range()?,
272            session_sk: config.session_sk.clone(),
273            storage: None,
274            measure: None,
275            stabilize_interval: config.stabilize_interval,
276            dht_finger_table_size: DEFAULT_FINGER_TABLE_SIZE,
277            reassembly_limits: ReassemblyLimits::production(),
278        })
279    }
280
281    /// Set the storage for the processor.
282    pub fn storage(mut self, storage: EntryStorage) -> Self {
283        self.storage = Some(storage);
284        self
285    }
286
287    /// Set the measure for the processor.
288    pub fn measure(mut self, implement: PeriodicMeasure) -> Self {
289        self.measure = Some(Arc::new(implement));
290        self
291    }
292
293    /// Set the number of DHT finger-table slots for the processor's swarm.
294    pub fn dht_finger_table_size(mut self, size: usize) -> Self {
295        self.dht_finger_table_size = size;
296        self
297    }
298
299    /// Set inbound chunk reassembly limits for the processor's swarm.
300    pub fn reassembly_limits(mut self, limits: ReassemblyLimits) -> Self {
301        self.reassembly_limits = limits;
302        self
303    }
304
305    /// Build the [Processor].
306    pub fn build(self) -> Result<Processor> {
307        self.session_sk
308            .session()
309            .verify_self()
310            .map_err(|e| Error::VerifyError(e.to_string()))?;
311
312        let storage = self.storage.unwrap_or_else(|| Box::new(MemStorage::new()));
313
314        let mut swarm_builder =
315            SwarmBuilder::new(self.network_id, &self.ice_servers, storage, self.session_sk);
316        swarm_builder = swarm_builder.dht_storage_redundancy(DATA_REDUNDANT);
317        swarm_builder = swarm_builder.dht_finger_table_size(self.dht_finger_table_size);
318        swarm_builder = swarm_builder.reassembly_limits(self.reassembly_limits);
319
320        if let Some(external_address) = self.external_address {
321            swarm_builder = swarm_builder.external_address(external_address);
322        }
323        if let Some(range) = self.webrtc_udp_port_range {
324            swarm_builder = swarm_builder.webrtc_udp_port_range(range);
325        }
326
327        if let Some(measure) = self.measure {
328            swarm_builder = swarm_builder.measure(measure);
329        }
330        let swarm = Arc::new(swarm_builder.build());
331
332        Ok(Processor {
333            swarm,
334            stabilize_interval: self.stabilize_interval,
335        })
336    }
337}
338
339impl Processor {
340    /// Get current did
341    pub fn did(&self) -> Did {
342        self.swarm.did()
343    }
344
345    /// Run stabilization daemon
346    pub async fn listen(&self) {
347        let stabilizer = self.swarm.stabilizer();
348        Arc::new(stabilizer).wait(self.stabilize_interval).await
349    }
350
351    /// Connect peer with web3 did.
352    /// There are 3 peers: PeerA, PeerB, PeerC.
353    /// 1. PeerA has a connection with PeerB.
354    /// 2. PeerC has a connection with PeerB.
355    /// 3. PeerC can connect PeerA with PeerA's web3 address.
356    pub async fn connect_with_did(&self, did: Did) -> Result<()> {
357        self.swarm.connect(did).await.map_err(Error::ConnectError)?;
358        Ok(())
359    }
360
361    /// Disconnect a peer with web3 did.
362    pub async fn disconnect(&self, did: Did) -> Result<()> {
363        self.swarm
364            .disconnect(did)
365            .await
366            .map_err(Error::CloseConnectionError)
367    }
368
369    /// Send custom message to a did.
370    pub async fn send_message(&self, destination: Did, msg: &[u8]) -> Result<uuid::Uuid> {
371        tracing::info!("send_message, message size: {:?}", msg.len());
372
373        let msg = Message::custom(msg).map_err(Error::SendMessage)?;
374
375        self.swarm
376            .send_message(msg, destination)
377            .await
378            .map_err(Error::SendMessage)
379    }
380
381    /// Send an E2E handshake request to a DID.
382    ///
383    /// The negotiated key is the peer's account/identity secp256k1 key, not
384    /// the ephemeral session key.
385    pub async fn send_e2e_handshake(&self, destination: Did) -> Result<uuid::Uuid> {
386        let public_key = self.swarm.account_pubkey().map_err(Error::SendMessage)?;
387        self.swarm
388            .send_message(
389                Message::E2eHandshakeRequest(E2eHandshakeRequest::new(public_key)),
390                destination,
391            )
392            .await
393            .map_err(Error::SendMessage)
394    }
395
396    /// Send an ElGamal-encrypted E2E message to a DID with a verified recipient key.
397    ///
398    /// Returns the stream id shared by all emitted E2E stream frames.
399    pub async fn send_e2e_message(
400        &self,
401        destination: Did,
402        recipient_public_key: PublicKey<33>,
403        msg: &[u8],
404    ) -> Result<uuid::Uuid> {
405        self.send_e2e_message_with_frame_len(
406            destination,
407            recipient_public_key,
408            msg,
409            e2e::DEFAULT_E2E_PLAINTEXT_FRAME_LEN,
410        )
411        .await
412    }
413
414    /// Send an ElGamal-encrypted E2E stream with an explicit plaintext frame size.
415    ///
416    /// Returns the stream id shared by all emitted E2E stream frames.
417    pub async fn send_e2e_message_with_frame_len(
418        &self,
419        destination: Did,
420        recipient_public_key: PublicKey<33>,
421        msg: &[u8],
422        max_plaintext_frame_len: usize,
423    ) -> Result<uuid::Uuid> {
424        e2e::ensure_public_key_matches_did(recipient_public_key, destination)
425            .map_err(Error::SendMessage)?;
426        let sender_public_key = self.swarm.account_pubkey().map_err(Error::SendMessage)?;
427        let stream_id = uuid::Uuid::new_v4();
428        let frames = e2e::encrypt_stream_frames(
429            msg,
430            stream_id,
431            sender_public_key,
432            recipient_public_key,
433            max_plaintext_frame_len,
434        )
435        .map_err(Error::SendMessage)?
436        .collect::<rings_core::error::Result<Vec<_>>>()
437        .map_err(Error::SendMessage)?;
438
439        for frame in frames {
440            self.swarm
441                .send_message(Message::E2eStreamFrame(frame), destination)
442                .await
443                .map_err(Error::SendMessage)?;
444        }
445
446        Ok(stream_id)
447    }
448
449    /// Verify an E2E handshake request and return the requester's identity public key.
450    pub fn verify_e2e_handshake_request(
451        &self,
452        requester: Did,
453        request: &E2eHandshakeRequest,
454    ) -> Result<PublicKey<33>> {
455        request
456            .verify_requester(requester)
457            .map_err(Error::CoreError)?;
458        Ok(request.requester_public_key)
459    }
460
461    /// Verify an E2E handshake response and return the responder's identity public key.
462    pub fn verify_e2e_handshake_response(
463        &self,
464        responder: Did,
465        response: &E2eHandshakeResponse,
466    ) -> Result<PublicKey<33>> {
467        response
468            .verify_responder(responder)
469            .map_err(Error::CoreError)?;
470        Ok(response.responder_public_key)
471    }
472
473    /// Create an E2E stream decryptor with this node's identity/signing secret key.
474    ///
475    /// The ciphertext is encrypted to the DID/account key negotiated by the
476    /// handshake. A session private key cannot decrypt it unless the session key
477    /// is also the account key, so callers must supply the local identity key
478    /// explicitly.
479    pub fn e2e_stream_decryptor(
480        &self,
481        expected_sender: Did,
482        stream_id: e2e::E2eStreamId,
483        recipient_identity_key: SecretKey,
484    ) -> Result<E2eStreamDecryptor> {
485        e2e::ensure_public_key_matches_did(recipient_identity_key.pubkey(), self.did())
486            .map_err(Error::CoreError)?;
487        Ok(E2eStreamDecryptor::new(
488            stream_id,
489            expected_sender,
490            recipient_identity_key,
491        ))
492    }
493
494    /// Decrypt one E2E stream frame with an already-created stream decryptor.
495    pub fn decrypt_e2e_stream_frame(
496        &self,
497        decryptor: &mut E2eStreamDecryptor,
498        frame: &E2eStreamFrame,
499    ) -> Result<Vec<u8>> {
500        decryptor.decrypt_next(frame).map_err(Error::CoreError)
501    }
502
503    /// Send a namespaced [`Envelope`](crate::extension::ext::Envelope) to a did over the
504    /// P2P transport (the wire codec
505    /// of the extension layer). `send_envelope : (Did, Envelope) → IO TxId`.
506    pub async fn send_envelope(
507        &self,
508        destination: Did,
509        envelope: &crate::extension::ext::Envelope,
510    ) -> Result<uuid::Uuid> {
511        let msg_bytes = envelope.encode()?;
512        self.send_message(destination, &msg_bytes).await
513    }
514
515    /// check local cache of dht
516    pub async fn storage_check_cache(&self, entry_key: Did) -> Option<entry::Entry> {
517        self.swarm.storage_check_cache(entry_key).await
518    }
519
520    /// Fetch an entry from DHT storage
521    pub async fn storage_fetch(&self, entry_key: Did) -> Result<()> {
522        <Swarm as ChordStorageInterface<DATA_REDUNDANT>>::storage_fetch(&self.swarm, entry_key)
523            .await
524            .map_err(Error::EntryError)
525    }
526
527    /// Store an entry on DHT storage
528    pub async fn storage_store(&self, entry: entry::Entry) -> Result<()> {
529        <Swarm as ChordStorageInterface<DATA_REDUNDANT>>::storage_store(&self.swarm, entry)
530            .await
531            .map_err(Error::EntryError)
532    }
533
534    /// Append data to an entry on DHT storage
535    pub async fn storage_append_data(&self, topic: &str, data: Encoded) -> Result<()> {
536        <Swarm as ChordStorageInterface<DATA_REDUNDANT>>::storage_append_data(
537            &self.swarm,
538            topic,
539            data,
540        )
541        .await
542        .map_err(Error::EntryError)
543    }
544
545    /// register service
546    pub async fn register_service(&self, name: &str) -> Result<()> {
547        let encoded_did = self
548            .did()
549            .to_string()
550            .encode()
551            .map_err(Error::ServiceRegisterError)?;
552        <Swarm as ChordStorageInterface<DATA_REDUNDANT>>::storage_touch_data(
553            &self.swarm,
554            name,
555            encoded_did,
556        )
557        .await
558        .map_err(Error::ServiceRegisterError)
559    }
560
561    /// get node info
562    pub async fn get_node_info(&self) -> Result<NodeInfoResponse> {
563        Ok(NodeInfoResponse {
564            version: crate::util::build_version(),
565            swarm: Some(self.swarm.inspect().await.into()),
566        })
567    }
568}
569
570#[cfg(test)]
571#[cfg(feature = "node")]
572mod test {
573    use std::sync::Mutex;
574    use std::time::Duration;
575    use std::time::Instant;
576
577    use rings_core::storage::MemStorage;
578    use rings_core::swarm::callback::SwarmCallback;
579    use rings_core::swarm::callback::SwarmEvent;
580    use rings_transport::core::transport::WebrtcConnectionState;
581    use tokio::sync::Notify;
582
583    use super::*;
584    use crate::prelude::*;
585    use crate::tests::native::prepare_processor;
586
587    #[test]
588    fn webrtc_udp_port_range_absent_by_default() {
589        let range = parse_webrtc_udp_port_range(None, None);
590
591        assert!(matches!(range, core::result::Result::Ok(None)));
592    }
593
594    #[test]
595    fn webrtc_udp_port_range_accepts_valid_bounds() {
596        let range = parse_webrtc_udp_port_range(Some(49160), Some(49200));
597
598        assert!(matches!(
599            range,
600            Ok(Some(range)) if range.min() == 49160 && range.max() == 49200
601        ));
602    }
603
604    #[test]
605    fn webrtc_udp_port_range_rejects_partial_bounds() {
606        let range = parse_webrtc_udp_port_range(Some(49160), None);
607
608        assert!(matches!(
609            range,
610            Err(Error::IncompleteWebrtcUdpPortRange {
611                min: Some(49160),
612                max: None
613            })
614        ));
615    }
616
617    #[test]
618    fn webrtc_udp_port_range_rejects_zero_bound() {
619        let range = parse_webrtc_udp_port_range(Some(0), Some(49200));
620
621        assert!(matches!(
622            range,
623            Err(Error::InvalidWebrtcUdpPortRange(
624                rings_transport::webrtc_config::WebrtcUdpPortRangeError::ZeroBound {
625                    min: 0,
626                    max: 49200
627                }
628            ))
629        ));
630    }
631
632    #[test]
633    fn webrtc_udp_port_range_rejects_inverted_bounds() {
634        let range = parse_webrtc_udp_port_range(Some(49200), Some(49160));
635
636        assert!(matches!(
637            range,
638            Err(Error::InvalidWebrtcUdpPortRange(
639                rings_transport::webrtc_config::WebrtcUdpPortRangeError::Inverted {
640                    min: 49200,
641                    max: 49160
642                }
643            ))
644        ));
645    }
646
647    #[tokio::test]
648    async fn test_processor_create_offer() {
649        let peer_did = SecretKey::random().address().into();
650        let processor = prepare_processor().await;
651        processor.swarm.create_offer(peer_did).await.unwrap();
652        let conn_dids = processor.swarm.peers();
653        assert_eq!(conn_dids.len(), 1);
654        assert_eq!(conn_dids.first().unwrap().did, peer_did.to_string());
655    }
656
657    struct SwarmCallbackInstance {
658        inbound: Mutex<Vec<Message>>,
659        inbound_notify: Notify,
660        connected_notify: Notify,
661    }
662
663    #[async_trait]
664    impl SwarmCallback for SwarmCallbackInstance {
665        async fn on_inbound(
666            &self,
667            payload: &MessagePayload,
668        ) -> std::result::Result<(), Box<dyn std::error::Error>> {
669            let msg: Message = payload.transaction.data().map_err(Box::new)?;
670            {
671                let mut inbound = self.inbound.lock().unwrap();
672                inbound.push(msg);
673            }
674            self.inbound_notify.notify_one();
675
676            Ok(())
677        }
678
679        async fn on_event(
680            &self,
681            event: &SwarmEvent,
682        ) -> std::result::Result<(), Box<dyn std::error::Error>> {
683            if let SwarmEvent::ConnectionStateChange {
684                state: WebrtcConnectionState::Connected,
685                ..
686            } = event
687            {
688                self.connected_notify.notify_one();
689            }
690
691            Ok(())
692        }
693    }
694
695    fn test_callback() -> Arc<SwarmCallbackInstance> {
696        Arc::new(SwarmCallbackInstance {
697            inbound: Mutex::new(Vec::new()),
698            inbound_notify: Notify::new(),
699            connected_notify: Notify::new(),
700        })
701    }
702
703    async fn prepare_processor_with_identity_key(identity_key: SecretKey) -> Processor {
704        let session_sk = SessionSk::new_with_seckey(&identity_key).unwrap();
705        let config = ProcessorConfig::new(
706            0,
707            "stun://stun.l.google.com:19302".to_string(),
708            session_sk,
709            3,
710        );
711        let storage = Box::new(MemStorage::new());
712
713        ProcessorBuilder::from_config(&config)
714            .unwrap()
715            .storage(storage)
716            .dht_finger_table_size(8)
717            .build()
718            .unwrap()
719    }
720
721    async fn connect_processors(
722        p1: &Processor,
723        p2: &Processor,
724        callback1: &SwarmCallbackInstance,
725        callback2: &SwarmCallbackInstance,
726    ) {
727        let offer = p1.swarm.create_offer(p2.did()).await.unwrap();
728        let answer = p2.swarm.answer_offer(offer).await.unwrap();
729        p1.swarm.accept_answer(answer).await.unwrap();
730        wait_processors_connected(p1, p2, callback1, callback2).await;
731    }
732
733    async fn wait_processors_connected(
734        p1: &Processor,
735        p2: &Processor,
736        callback1: &SwarmCallbackInstance,
737        callback2: &SwarmCallbackInstance,
738    ) {
739        let deadline = Instant::now() + Duration::from_secs(5);
740        loop {
741            if processor_has_connected_peer(p1, p2.did())
742                && processor_has_connected_peer(p2, p1.did())
743            {
744                return;
745            }
746
747            let remaining = deadline
748                .checked_duration_since(Instant::now())
749                .expect("processors did not connect");
750            tokio::time::timeout(remaining, async {
751                tokio::select! {
752                    _ = callback1.connected_notify.notified() => {}
753                    _ = callback2.connected_notify.notified() => {}
754                }
755            })
756            .await
757            .expect("processors did not connect");
758        }
759    }
760
761    fn processor_has_connected_peer(processor: &Processor, peer: Did) -> bool {
762        let peer = peer.to_string();
763        processor
764            .swarm
765            .peers()
766            .into_iter()
767            .any(|conn| conn.did == peer && conn.state == "Connected")
768    }
769
770    async fn wait_for_inbound_message(
771        callback: &SwarmCallbackInstance,
772        predicate: impl Fn(&Message) -> bool,
773    ) -> Message {
774        let deadline = Instant::now() + Duration::from_secs(5);
775        loop {
776            {
777                let inbound = callback.inbound.lock().unwrap();
778                if let Some(msg) = inbound.iter().find(|msg| predicate(msg)).cloned() {
779                    return msg;
780                }
781            }
782
783            let remaining = deadline
784                .checked_duration_since(Instant::now())
785                .expect("inbound message was not delivered");
786            tokio::time::timeout(remaining, callback.inbound_notify.notified())
787                .await
788                .expect("inbound message was not delivered");
789        }
790    }
791
792    async fn wait_for_e2e_stream_frames(
793        callback: &SwarmCallbackInstance,
794        stream_id: e2e::E2eStreamId,
795    ) -> Vec<E2eStreamFrame> {
796        let deadline = Instant::now() + Duration::from_secs(5);
797        loop {
798            {
799                let inbound = callback.inbound.lock().unwrap();
800                let frames = inbound
801                    .iter()
802                    .filter_map(|msg| match msg {
803                        Message::E2eStreamFrame(frame) if frame.stream_id == stream_id => {
804                            Some(frame.clone())
805                        }
806                        _ => None,
807                    })
808                    .collect::<Vec<_>>();
809                if frames.iter().any(|frame| frame.is_final) {
810                    return frames;
811                }
812            }
813
814            let remaining = deadline
815                .checked_duration_since(Instant::now())
816                .expect("E2E stream final frame was not delivered");
817            tokio::time::timeout(remaining, callback.inbound_notify.notified())
818                .await
819                .expect("E2E stream final frame was not delivered");
820        }
821    }
822
823    #[tokio::test]
824    async fn test_processor_handshake_msg() {
825        let callback1 = test_callback();
826        let callback2 = test_callback();
827
828        let p1 = prepare_processor().await;
829        let p2 = prepare_processor().await;
830
831        p1.swarm.set_callback(callback1.clone()).unwrap();
832        p2.swarm.set_callback(callback2.clone()).unwrap();
833
834        let did1 = p1.did();
835        let did2 = p2.did();
836
837        let offer = p1.swarm.create_offer(p2.did()).await.unwrap();
838        assert_eq!(
839            p1.swarm
840                .peers()
841                .into_iter()
842                .find(|peer| peer.did == p2.did().to_string())
843                .unwrap()
844                .state,
845            "New"
846        );
847
848        let answer = p2.swarm.answer_offer(offer).await.unwrap();
849        p1.swarm.accept_answer(answer).await.unwrap();
850        wait_processors_connected(&p1, &p2, &callback1, &callback2).await;
851
852        let test_text1 = "test1";
853        let test_text2 = "test2";
854
855        p1.send_message(did2, test_text1.as_bytes()).await.unwrap();
856        p2.send_message(did1, test_text2.as_bytes()).await.unwrap();
857
858        let got_msg2 = wait_for_inbound_message(&callback2, |msg| {
859            matches!(msg, Message::CustomMessage(custom) if custom.0 == test_text1.as_bytes())
860        })
861        .await;
862        assert!(matches!(got_msg2, Message::CustomMessage(_)));
863
864        let got_msg1 = wait_for_inbound_message(&callback1, |msg| {
865            matches!(msg, Message::CustomMessage(custom) if custom.0 == test_text2.as_bytes())
866        })
867        .await;
868        assert!(matches!(got_msg1, Message::CustomMessage(_)));
869    }
870
871    #[tokio::test]
872    async fn test_processor_e2e_handshake_exchanges_verified_public_keys() {
873        let callback1 = test_callback();
874        let callback2 = test_callback();
875
876        let p1 = prepare_processor().await;
877        let p2 = prepare_processor().await;
878
879        p1.swarm.set_callback(callback1.clone()).unwrap();
880        p2.swarm.set_callback(callback2.clone()).unwrap();
881
882        connect_processors(&p1, &p2, &callback1, &callback2).await;
883
884        let did1 = p1.did();
885        let did2 = p2.did();
886        let requester_public_key = p1.swarm.account_pubkey().unwrap();
887        let responder_public_key = p2.swarm.account_pubkey().unwrap();
888
889        p1.send_e2e_handshake(did2).await.unwrap();
890
891        let request = wait_for_inbound_message(&callback2, |msg| {
892            matches!(msg, Message::E2eHandshakeRequest(_))
893        })
894        .await;
895        match request {
896            Message::E2eHandshakeRequest(request) => {
897                assert_eq!(request.requester_public_key, requester_public_key);
898                assert_eq!(
899                    p2.verify_e2e_handshake_request(did1, &request).unwrap(),
900                    requester_public_key
901                );
902            }
903            msg => panic!("expected E2eHandshakeRequest, got {msg:?}"),
904        }
905
906        let response = wait_for_inbound_message(&callback1, |msg| {
907            matches!(msg, Message::E2eHandshakeResponse(_))
908        })
909        .await;
910        match response {
911            Message::E2eHandshakeResponse(response) => {
912                assert_eq!(response.responder_public_key, responder_public_key);
913                assert_eq!(
914                    p1.verify_e2e_handshake_response(did2, &response).unwrap(),
915                    responder_public_key
916                );
917            }
918            msg => panic!("expected E2eHandshakeResponse, got {msg:?}"),
919        }
920    }
921
922    #[tokio::test]
923    async fn test_processor_e2e_message_streams_and_decrypts_with_receiver_identity_key() {
924        let callback1 = test_callback();
925        let callback2 = test_callback();
926        let identity1 = SecretKey::random();
927        let identity2 = SecretKey::random();
928
929        let p1 = prepare_processor_with_identity_key(identity1).await;
930        let p2 = prepare_processor_with_identity_key(identity2).await;
931
932        p1.swarm.set_callback(callback1.clone()).unwrap();
933        p2.swarm.set_callback(callback2.clone()).unwrap();
934
935        connect_processors(&p1, &p2, &callback1, &callback2).await;
936
937        let did1 = p1.did();
938        let did2 = p2.did();
939        let responder_public_key = p2.swarm.account_pubkey().unwrap();
940        let stream_id = p1
941            .send_e2e_message_with_frame_len(
942                did2,
943                responder_public_key,
944                b"homomorphic-ready streaming body",
945                8,
946            )
947            .await
948            .unwrap();
949
950        let frames = wait_for_e2e_stream_frames(&callback2, stream_id).await;
951        assert!(
952            frames.len() > 1,
953            "streaming send should emit more than one frame for this frame size"
954        );
955        assert_eq!(
956            frames.iter().filter(|frame| frame.is_final).count(),
957            1,
958            "streaming send should emit exactly one final frame"
959        );
960
961        let mut sequences = frames
962            .iter()
963            .map(|frame| frame.sequence)
964            .collect::<Vec<_>>();
965        sequences.sort_unstable();
966        let frame_count = u64::try_from(frames.len()).unwrap();
967        assert_eq!(sequences, (0..frame_count).collect::<Vec<_>>());
968
969        let mut decryptor = p2.e2e_stream_decryptor(did1, stream_id, identity2).unwrap();
970        let mut plaintext = Vec::new();
971        let mut delivered_frames = frames.clone();
972        delivered_frames.reverse();
973        for frame in &delivered_frames {
974            plaintext
975                .extend_from_slice(&p2.decrypt_e2e_stream_frame(&mut decryptor, frame).unwrap());
976        }
977        decryptor.finish().unwrap();
978        assert_eq!(plaintext, b"homomorphic-ready streaming body");
979
980        assert!(matches!(
981            p2.e2e_stream_decryptor(did1, stream_id, SecretKey::random()),
982            Err(Error::CoreError(
983                rings_core::error::Error::E2ePublicKeyDidMismatch { .. }
984            ))
985        ));
986    }
987}
rings_node/processor.rs

rings_node/
processor.rs
