citadel_sdk/
remote_ext.rs

1//! Remote Protocol Extensions
2//!
3//! This module extends the core NodeRemote functionality with high-level operations
4//! for managing connections, file transfers, and peer interactions in the Citadel
5//! Protocol network.
6//!
7//! # Features
8//! - User registration and authentication
9//! - Connection management
10//! - File transfer operations
11//! - Virtual filesystem support
12//! - Peer discovery and management
13//! - Group communication
14//! - Security settings configuration
15//!
16//! # Example
17//! ```rust
18//! use citadel_sdk::prelude::*;
19//!
20//! async fn example<R: Ratchet>(remote: NodeRemote<R>) -> Result<(), NetworkError> {
21//!     // Register a new user
22//!     let reg = remote.register_with_defaults(
23//!         "127.0.0.1:25021",
24//!         "John Doe",
25//!         "john.doe",
26//!         "password123"
27//!     ).await?;
28//!
29//!     // Connect to a peer
30//!     let auth = AuthenticationRequest::credentialed("john.doe", "password123");
31//!     let conn = remote.connect_with_defaults(auth).await?;
32//!
33//!     // Send a file to a peer
34//!     remote.find_target("john.doe", "peer.name")
35//!         .await?
36//!         .send_file("/path/to/file.txt")
37//!         .await?;
38//!
39//!     Ok(())
40//! }
41//! ```
42//!
43//! # Important Notes
44//! - All operations are asynchronous
45//! - Connections are automatically managed
46//! - File transfers support chunking
47//! - Virtual filesystem is encrypted
48//! - Peer connections require mutual registration
49//!
50//! # Related Components
51//! - [`NodeRemote`]: Core remote interface
52//! - [`ClientServerRemote`]: Client-server communication
53//! - [`PeerRemote`]: Peer-to-peer communication
54//! - [`CitadelClientServerConnection`]: Connection management
55//! - [`RegisterSuccess`]: Registration handling
56//!
57//! [`NodeRemote`]: crate::prelude::NodeRemote
58//! [`ClientServerRemote`]: crate::prelude::ClientServerRemote
59//! [`PeerRemote`]: crate::prelude::PeerRemote
60//! [`CitadelClientServerConnection`]: crate::prelude::CitadelClientServerConnection
61//! [`RegisterSuccess`]: crate::prelude::RegisterSuccess
62
63use crate::prefabs::ClientServerRemote;
64use crate::prelude::results::{PeerConnectSuccess, PeerRegisterStatus};
65use crate::prelude::*;
66use crate::remote_ext::remote_specialization::PeerRemote;
67use crate::remote_ext::results::LocalGroupPeerFullInfo;
68use std::ops::{Deref, DerefMut};
69
70use futures::StreamExt;
71use std::path::PathBuf;
72use std::time::Duration;
73
74pub(crate) mod user_ids {
75    use crate::prelude::*;
76    use std::ops::Deref;
77
78    #[derive(Debug)]
79    /// A reference to a user identifier
80    pub struct SymmetricIdentifierHandleRef<'a, R: Ratchet> {
81        pub(crate) user: VirtualTargetType,
82        pub(crate) remote: &'a NodeRemote<R>,
83        pub(crate) target_username: Option<String>,
84    }
85
86    impl<R: Ratchet> SymmetricIdentifierHandleRef<'_, R> {
87        pub fn into_owned(self) -> SymmetricIdentifierHandle<R> {
88            SymmetricIdentifierHandle {
89                user: self.user,
90                remote: self.remote.clone(),
91                target_username: self.target_username,
92            }
93        }
94    }
95
96    #[derive(Clone, Debug)]
97    /// A convenience structure for executing commands that depend on a specific registered user
98    pub struct SymmetricIdentifierHandle<R: Ratchet> {
99        user: VirtualTargetType,
100        remote: NodeRemote<R>,
101        target_username: Option<String>,
102    }
103
104    pub trait TargetLockedRemote<R: Ratchet>: Send + Sync {
105        fn user(&self) -> &VirtualTargetType;
106        fn remote(&self) -> &NodeRemote<R>;
107        fn target_username(&self) -> Option<&str>;
108        fn user_mut(&mut self) -> &mut VirtualTargetType;
109        fn session_security_settings(&self) -> Option<&SessionSecuritySettings>;
110    }
111
112    impl<R: Ratchet> TargetLockedRemote<R> for SymmetricIdentifierHandleRef<'_, R> {
113        fn user(&self) -> &VirtualTargetType {
114            &self.user
115        }
116        fn remote(&self) -> &NodeRemote<R> {
117            self.remote
118        }
119        fn target_username(&self) -> Option<&str> {
120            self.target_username.as_deref()
121        }
122        fn user_mut(&mut self) -> &mut VirtualTargetType {
123            &mut self.user
124        }
125
126        fn session_security_settings(&self) -> Option<&SessionSecuritySettings> {
127            None
128        }
129    }
130
131    impl<R: Ratchet> TargetLockedRemote<R> for SymmetricIdentifierHandle<R> {
132        fn user(&self) -> &VirtualTargetType {
133            &self.user
134        }
135        fn remote(&self) -> &NodeRemote<R> {
136            &self.remote
137        }
138        fn target_username(&self) -> Option<&str> {
139            self.target_username.as_deref()
140        }
141        fn user_mut(&mut self) -> &mut VirtualTargetType {
142            &mut self.user
143        }
144
145        fn session_security_settings(&self) -> Option<&SessionSecuritySettings> {
146            None
147        }
148    }
149
150    impl<R: Ratchet> From<SymmetricIdentifierHandleRef<'_, R>> for SymmetricIdentifierHandle<R> {
151        fn from(this: SymmetricIdentifierHandleRef<'_, R>) -> Self {
152            this.into_owned()
153        }
154    }
155
156    impl<R: Ratchet> Deref for SymmetricIdentifierHandle<R> {
157        type Target = NodeRemote<R>;
158
159        fn deref(&self) -> &Self::Target {
160            &self.remote
161        }
162    }
163
164    impl<R: Ratchet> Deref for SymmetricIdentifierHandleRef<'_, R> {
165        type Target = NodeRemote<R>;
166
167        fn deref(&self) -> &Self::Target {
168            self.remote
169        }
170    }
171}
172
173/// Contains the elements required to communicate with the adjacent node
174pub struct CitadelClientServerConnection<R: Ratchet> {
175    /// An interface to send ordered, reliable, and encrypted messages
176    pub(crate) channel: Option<PeerChannel<R>>,
177    pub remote: ClientServerRemote<R>,
178    /// Only available if UdpMode was enabled at the beginning of a session
179    pub udp_channel_rx: Option<citadel_io::tokio::sync::oneshot::Receiver<UdpChannel<R>>>,
180    /// Contains the Google auth minted at the central server (if the central server enabled it), as well as any other services enabled by the central server
181    pub services: ServicesObject,
182    pub cid: u64,
183    pub session_security_settings: SessionSecuritySettings,
184}
185
186impl<R: Ratchet> CitadelClientServerConnection<R> {
187    /// Splits the channel into a send and receive half. This will render
188    /// the other fields of this connection object innaccessible
189    ///
190    /// # Panics
191    ///  - If the channel has already been taken
192    pub fn split(self) -> (PeerChannelSendHalf<R>, PeerChannelRecvHalf<R>) {
193        self.channel.expect("Channel already taken").split()
194    }
195
196    pub fn take_channel(&mut self) -> Option<PeerChannel<R>> {
197        self.channel.take()
198    }
199}
200
201impl<R: Ratchet> Deref for CitadelClientServerConnection<R> {
202    type Target = ClientServerRemote<R>;
203
204    fn deref(&self) -> &Self::Target {
205        &self.remote
206    }
207}
208
209impl<R: Ratchet> DerefMut for CitadelClientServerConnection<R> {
210    fn deref_mut(&mut self) -> &mut Self::Target {
211        &mut self.remote
212    }
213}
214
215/// Contains the elements entailed by a successful registration
216pub struct RegisterSuccess {
217    pub cid: u64,
218}
219
220/// Waits for the outcome of a registration request.
221async fn await_registration<R: Ratchet, Rem: Remote<R>>(
222    remote: &Rem,
223    register_request: NodeRequest,
224) -> Result<RegisterSuccess, NetworkError> {
225    let mut subscription = remote.send_callback_subscription(register_request).await?;
226    while let Some(status) = subscription.next().await {
227        match status.into_result()? {
228            NodeResult::RegisterOkay(RegisterOkay { cid, .. }) => {
229                return Ok(RegisterSuccess { cid });
230            }
231            NodeResult::RegisterFailure(err) => {
232                return Err(citadel_io::error!(
233                    citadel_io::ErrorCode::RemoteRegisterFailure,
234                    err.error_message
235                ));
236            }
237            NodeResult::Disconnect(err) => {
238                return Err(citadel_io::error!(
239                    citadel_io::ErrorCode::RemoteDisconnected,
240                    err.message
241                ));
242            }
243            evt => {
244                log::warn!(target: "citadel", "Invalid NodeResult for Register request received: {evt:?}");
245            }
246        }
247    }
248
249    Err(citadel_io::error!(
250        citadel_io::ErrorCode::RemoteKernelStreamDied,
251        "register"
252    ))
253}
254
255#[async_trait]
256/// Endows the [NodeRemote](NodeRemote) with additional functions
257pub trait ProtocolRemoteExt<R: Ratchet>: Remote<R> {
258    /// Registers with custom settings
259    /// Returns a ticket which is used to uniquely identify the request in the protocol
260    async fn register<
261        T: std::net::ToSocketAddrs + Send,
262        P: Into<String> + Send,
263        V: Into<String> + Send,
264        K: Into<SecBuffer> + Send,
265    >(
266        &self,
267        addr: T,
268        full_name: P,
269        username: V,
270        proposed_password: K,
271        default_security_settings: SessionSecuritySettings,
272        server_password: Option<PreSharedKey>,
273    ) -> Result<RegisterSuccess, NetworkError> {
274        let creds =
275            ProposedCredentials::new_register(full_name, username, proposed_password.into())
276                .await?;
277        let register_request = NodeRequest::RegisterToHypernode(RegisterToHypernode {
278            remote_addr: addr.to_socket_addrs()?.next().ok_or(citadel_io::error!(
279                citadel_io::ErrorCode::RemoteInvalidSocketAddr
280            ))?,
281            proposed_credentials: creds,
282            static_security_settings: default_security_settings,
283            session_password: server_password.unwrap_or_default(),
284            endpoint: None,
285        });
286
287        await_registration(self, register_request).await
288    }
289
290    /// Registers with a server reached by a WebSocket URL (`wss://org.example.net/`, or
291    /// `ws://host:port/path`) rather than at an address of its own: a server behind an HTTP edge,
292    /// which is told apart from the edge's other servers by hostname and path. The account
293    /// remembers the URL, so a later [`connect`](Self::connect) with credentials dials it too.
294    #[cfg(not(target_family = "wasm"))]
295    async fn register_to_endpoint<
296        P: Into<String> + Send,
297        V: Into<String> + Send,
298        K: Into<SecBuffer> + Send,
299    >(
300        &self,
301        endpoint: citadel_io::WebSocketEndpoint,
302        full_name: P,
303        username: V,
304        proposed_password: K,
305        default_security_settings: SessionSecuritySettings,
306        server_password: Option<PreSharedKey>,
307    ) -> Result<RegisterSuccess, NetworkError> {
308        let remote_addr = endpoint.resolve().await?;
309        let creds =
310            ProposedCredentials::new_register(full_name, username, proposed_password.into())
311                .await?;
312        let register_request = NodeRequest::RegisterToHypernode(RegisterToHypernode {
313            remote_addr,
314            proposed_credentials: creds,
315            static_security_settings: default_security_settings,
316            session_password: server_password.unwrap_or_default(),
317            endpoint: Some(endpoint),
318        });
319
320        await_registration(self, register_request).await
321    }
322
323    /// The WebSocket URL the local account `cid` registered to (see
324    /// [`register_to_endpoint`](Self::register_to_endpoint)), or `None` when its server is the
325    /// address in its connection info. Behind an HTTP edge that address is one of the edge's,
326    /// shared with every other server there, so it does not identify the server; this does.
327    async fn server_endpoint(
328        &self,
329        cid: u64,
330    ) -> Result<Option<citadel_io::WebSocketEndpoint>, NetworkError> {
331        citadel_proto::prelude::load_server_endpoint(self.account_manager(), cid).await
332    }
333
334    /// Registers using the default settings. The default uses No Google FCM keys and the default session security settings
335    /// Returns a ticket which is used to uniquely identify the request in the protocol
336    async fn register_with_defaults<
337        T: std::net::ToSocketAddrs + Send,
338        P: Into<String> + Send,
339        V: Into<String> + Send,
340        K: Into<SecBuffer> + Send,
341    >(
342        &self,
343        addr: T,
344        full_name: P,
345        username: V,
346        proposed_password: K,
347    ) -> Result<RegisterSuccess, NetworkError> {
348        self.register(
349            addr,
350            full_name,
351            username,
352            proposed_password,
353            Default::default(),
354            Default::default(),
355        )
356        .await
357    }
358
359    /// Connects with custom settings
360    /// Returns a ticket which is used to uniquely identify the request in the protocol
361    async fn connect(
362        &self,
363        auth: AuthenticationRequest,
364        connect_mode: ConnectMode,
365        udp_mode: UdpMode,
366        keep_alive_timeout: Option<Duration>,
367        session_security_settings: SessionSecuritySettings,
368        server_password: Option<PreSharedKey>,
369    ) -> Result<CitadelClientServerConnection<R>, NetworkError> {
370        let connect_request = NodeRequest::ConnectToHypernode(ConnectToHypernode {
371            auth_request: auth,
372            connect_mode,
373            udp_mode,
374            keep_alive_timeout: keep_alive_timeout.map(|r| r.as_secs()),
375            session_security_settings,
376            session_password: server_password.unwrap_or_default(),
377        });
378
379        let mut subscription = self.send_callback_subscription(connect_request).await?;
380        let status = subscription.next().await.ok_or(citadel_io::error!(
381            citadel_io::ErrorCode::RemoteKernelStreamDied,
382            "connect"
383        ))?;
384
385        return match status.into_result()? {
386            NodeResult::ConnectSuccess(ConnectSuccess {
387                ticket: _,
388                session_cid: cid,
389                remote_addr: _,
390                is_personal: _,
391                v_conn_type,
392                services,
393                welcome_message: _,
394                channel,
395                udp_rx_opt: udp_channel_rx,
396                session_security_settings,
397            }) => Ok(CitadelClientServerConnection {
398                remote: ClientServerRemote::new(
399                    v_conn_type,
400                    self.remote_ref().clone(),
401                    session_security_settings,
402                    None,
403                    None,
404                ),
405                channel: Some(*channel),
406                udp_channel_rx,
407                services,
408                cid,
409                session_security_settings,
410            }),
411            NodeResult::ConnectFail(ConnectFail {
412                ticket: _,
413                cid_opt: _,
414                error_message: err,
415            }) => Err(citadel_io::error!(
416                citadel_io::ErrorCode::RemoteConnectFailed,
417                err
418            )),
419            NodeResult::Disconnect(err) => {
420                return Err(citadel_io::error!(
421                    citadel_io::ErrorCode::RemoteDisconnected,
422                    err.message
423                ));
424            }
425            res => Err(citadel_io::error!(
426                citadel_io::ErrorCode::RemoteConnectUnexpectedResponse,
427                citadel_io::Dbg(res)
428            )),
429        };
430    }
431
432    /// Connects with the default settings
433    /// If FCM keys were created during the registration phase, then those keys will be used for the session. If new FCM keys need to be used, consider using [`Self::connect`]
434    async fn connect_with_defaults(
435        &self,
436        auth: AuthenticationRequest,
437    ) -> Result<CitadelClientServerConnection<R>, NetworkError> {
438        self.connect(
439            auth,
440            Default::default(),
441            Default::default(),
442            None,
443            Default::default(),
444            Default::default(),
445        )
446        .await
447    }
448
449    /// Creates a valid target identifier used to make protocol requests. Raw user IDs or usernames can be used
450    /// ```
451    /// use citadel_sdk::prelude::*;
452    /// # use citadel_sdk::prefabs::client::single_connection::SingleClientServerConnectionKernel;
453    ///
454    /// let server_connection_settings = DefaultServerConnectionSettingsBuilder::credentialed_login("127.0.0.1:25021", "john.doe", "password").build().unwrap();
455    ///
456    /// # SingleClientServerConnectionKernel::new(server_connection_settings, |conn| async move {
457    /// conn.find_target("my_account", "my_peer").await?.send_file("/path/to/file.pdf").await
458    /// // or: conn.find_target(1234, "my_peer").await? [...]
459    /// # });
460    /// ```
461    async fn find_target<T: Into<UserIdentifier> + Send, P: Into<UserIdentifier> + Send>(
462        &self,
463        local_user: T,
464        peer: P,
465    ) -> Result<SymmetricIdentifierHandleRef<'_, R>, NetworkError> {
466        let account_manager = self.account_manager();
467        account_manager
468            .find_target_information(local_user, peer)
469            .await?
470            .map(move |(cid, peer)| {
471                if peer.parent_icid != 0 {
472                    SymmetricIdentifierHandleRef {
473                        user: VirtualTargetType::ExternalGroupPeer {
474                            session_cid: cid,
475                            interserver_cid: peer.parent_icid,
476                            peer_cid: peer.cid,
477                        },
478                        remote: self.remote_ref(),
479                        target_username: None,
480                    }
481                } else {
482                    SymmetricIdentifierHandleRef {
483                        user: VirtualTargetType::LocalGroupPeer {
484                            session_cid: cid,
485                            peer_cid: peer.cid,
486                        },
487                        remote: self.remote_ref(),
488                        target_username: None,
489                    }
490                }
491            })
492            .ok_or_else(|| citadel_io::error!(citadel_io::ErrorCode::RemoteTargetPairNotFound))
493    }
494
495    /// Creates a proposed target from the valid local user to an unregistered peer in the network. Used when creating registration requests for peers.
496    /// Currently only supports LocalGroup <-> LocalGroup peer connections
497    async fn propose_target<T: Into<UserIdentifier> + Send, P: Into<UserIdentifier> + Send>(
498        &self,
499        local_user: T,
500        peer: P,
501    ) -> Result<SymmetricIdentifierHandleRef<'_, R>, NetworkError> {
502        let local_cid = self.get_session_cid(local_user).await?;
503        match peer.into() {
504            UserIdentifier::ID(peer_cid) => Ok(SymmetricIdentifierHandleRef {
505                user: VirtualTargetType::LocalGroupPeer {
506                    session_cid: local_cid,
507                    peer_cid,
508                },
509                remote: self.remote_ref(),
510                target_username: None,
511            }),
512            UserIdentifier::Username(uname) => {
513                let peer_cid = self
514                    .remote_ref()
515                    .account_manager()
516                    .find_target_information(local_cid, uname.clone())
517                    .await?
518                    .map(|r| r.1.cid)
519                    .unwrap_or(0);
520                Ok(SymmetricIdentifierHandleRef {
521                    user: VirtualTargetType::LocalGroupPeer {
522                        session_cid: local_cid,
523                        peer_cid,
524                    },
525                    remote: self.remote_ref(),
526                    target_username: Some(uname),
527                })
528            }
529        }
530    }
531
532    /// Returns a list of local group peers on the network for local_user. May or may not be registered to the user. To get a list of registered users to local_user, run [`Self::get_local_group_mutual_peers`]
533    /// - limit: if None, all peers are obtained. If Some, at most the specified number of peers will be obtained
534    async fn get_local_group_peers<T: Into<UserIdentifier> + Send>(
535        &self,
536        local_user: T,
537        limit: Option<usize>,
538    ) -> Result<Vec<LocalGroupPeerFullInfo>, NetworkError> {
539        let local_cid = self.get_session_cid(local_user).await?;
540        let command = NodeRequest::PeerCommand(PeerCommand {
541            session_cid: local_cid,
542            command: PeerSignal::GetRegisteredPeers {
543                peer_conn_type: ClientConnectionType::Server {
544                    session_cid: local_cid,
545                },
546                response: None,
547                limit: limit.map(|r| r as i32),
548            },
549        });
550
551        let mut stream = self.send_callback_subscription(command).await?;
552
553        while let Some(status) = stream.next().await {
554            if let NodeResult::PeerEvent(PeerEvent {
555                event:
556                    PeerSignal::GetRegisteredPeers {
557                        peer_conn_type: _,
558                        response: Some(PeerResponse::RegisteredCids(peer_info, is_onlines)),
559                        limit: _,
560                    },
561                ..
562            }) = status.into_result()?
563            {
564                return Ok(peer_info
565                    .into_iter()
566                    .zip(is_onlines)
567                    .filter_map(|(peer_info, is_online)| {
568                        peer_info.map(|info| LocalGroupPeerFullInfo {
569                            cid: info.cid,
570                            username: Some(info.username),
571                            full_name: Some(info.full_name),
572                            is_online,
573                        })
574                    })
575                    .collect());
576            }
577        }
578
579        Err(citadel_io::error!(
580            citadel_io::ErrorCode::RemoteKernelStreamDied,
581            "get_local_group_peers"
582        ))
583    }
584
585    /// Returns a list of mutually-registered peers with the local_user
586    async fn get_local_group_mutual_peers<T: Into<UserIdentifier> + Send>(
587        &self,
588        local_user: T,
589    ) -> Result<Vec<LocalGroupPeerFullInfo>, NetworkError> {
590        let local_cid = self.get_session_cid(local_user).await?;
591        let command = NodeRequest::PeerCommand(PeerCommand {
592            session_cid: local_cid,
593            command: PeerSignal::GetMutuals {
594                v_conn_type: ClientConnectionType::Server {
595                    session_cid: local_cid,
596                },
597                response: None,
598            },
599        });
600
601        let mut stream = self.send_callback_subscription(command).await?;
602
603        while let Some(status) = stream.next().await {
604            if let NodeResult::PeerEvent(PeerEvent {
605                event:
606                    PeerSignal::GetMutuals {
607                        v_conn_type: _,
608                        response: Some(PeerResponse::RegisteredCids(peer_info, is_onlines)),
609                    },
610                ..
611            }) = status.into_result()?
612            {
613                return Ok(peer_info
614                    .into_iter()
615                    .zip(is_onlines)
616                    .filter_map(|(peer_info, is_online)| {
617                        peer_info.map(|info| LocalGroupPeerFullInfo {
618                            cid: info.cid,
619                            username: Some(info.username),
620                            full_name: Some(info.full_name),
621                            is_online,
622                        })
623                    })
624                    .collect());
625            }
626        }
627
628        Err(citadel_io::error!(
629            citadel_io::ErrorCode::RemoteSessionStreamDied
630        ))
631    }
632
633    /// Returns all the active sessions in the protocol, including all P2P connections hierarchically placed as children to C2S
634    /// connections
635    async fn sessions(&self) -> Result<ActiveSessions, NetworkError> {
636        match self
637            .send_callback_subscription(NodeRequest::GetActiveSessions)
638            .await
639        {
640            Ok(mut stream) => {
641                while let Some(result) = stream.next().await {
642                    match result.into_result()? {
643                        NodeResult::SessionList(res) => return Ok(res.sessions),
644                        res => {
645                            citadel_logging::warn!("Received unexpected result: {res:?}");
646                        }
647                    }
648                }
649
650                citadel_logging::warn!("Failed to receive response from SDK (stream died)");
651                return Err(citadel_io::error!(
652                    citadel_io::ErrorCode::RemoteKernelStreamDied,
653                    "get_active_sessions"
654                ));
655            }
656            Err(e) => {
657                citadel_logging::warn!(target: "citadel", "Failed to query SDK sessions: {e}");
658            }
659        }
660
661        Err(citadel_io::error!(
662            citadel_io::ErrorCode::RemoteQuerySessionsFailed
663        ))
664    }
665
666    #[doc(hidden)]
667    fn remote_ref(&self) -> &NodeRemote<R>;
668
669    #[doc(hidden)]
670    async fn get_session_cid<T: Into<UserIdentifier> + Send>(
671        &self,
672        local_user: T,
673    ) -> Result<u64, NetworkError> {
674        let account_manager = self.account_manager();
675        Ok(account_manager
676            .find_local_user_information(local_user)
677            .await?
678            .ok_or(citadel_io::error!(
679                citadel_io::ErrorCode::RemoteUserDoesNotExist
680            ))?)
681    }
682}
683
684impl<R: Ratchet> ProtocolRemoteExt<R> for NodeRemote<R> {
685    fn remote_ref(&self) -> &NodeRemote<R> {
686        self
687    }
688}
689
690impl<R: Ratchet> ProtocolRemoteExt<R> for ClientServerRemote<R> {
691    fn remote_ref(&self) -> &NodeRemote<R> {
692        &self.inner
693    }
694}
695
696#[async_trait]
697/// Some functions require that a target exists
698pub trait ProtocolRemoteTargetExt<R: Ratchet>: TargetLockedRemote<R> {
699    /// Sends a file with a custom size. The smaller the chunks, the higher the degree of scrambling, but the higher the performance cost. A chunk size of zero will use the default
700    async fn send_file_with_custom_opts<T: ObjectSource>(
701        &self,
702        source: T,
703        chunk_size: usize,
704        transfer_type: TransferType,
705    ) -> Result<(), NetworkError> {
706        let chunk_size = if chunk_size == 0 {
707            None
708        } else {
709            Some(chunk_size)
710        };
711        let session_cid = self.user().get_session_cid();
712        let user = *self.user();
713        let remote = self.remote();
714
715        let mut stream = remote
716            .send_callback_subscription(NodeRequest::SendObject(SendObject {
717                source: Box::new(source),
718                chunk_size,
719                session_cid,
720                v_conn_type: user,
721                transfer_type,
722            }))
723            .await?;
724
725        while let Some(event) = stream.next().await {
726            match event.into_result()? {
727                NodeResult::ObjectTransferHandle(ObjectTransferHandle { mut handle, .. }) => {
728                    return handle.transfer_file().await.map_err(|err| {
729                        citadel_io::error!(
730                            citadel_io::ErrorCode::RemoteFileTransferFailed,
731                            err.into_string()
732                        )
733                    });
734                }
735
736                NodeResult::PeerEvent(PeerEvent {
737                    event: PeerSignal::SignalReceived { .. },
738                    ..
739                }) => {}
740
741                res => {
742                    log::warn!(target: "citadel", "Invalid NodeResult for FileTransfer request received: {res:?}")
743                }
744            }
745        }
746
747        Err(citadel_io::error!(
748            citadel_io::ErrorCode::RemoteFileTransferStreamDied
749        ))
750    }
751
752    /// Sends a file to the provided target using the default chunking size
753    async fn send_file<T: ObjectSource>(&self, source: T) -> Result<(), NetworkError> {
754        self.send_file_with_custom_opts(source, 0, TransferType::FileTransfer)
755            .await
756    }
757
758    /// Sends a file to the provided target using custom chunking size with local encryption.
759    /// Only this local node may decrypt the information send to the adjacent node.
760    async fn remote_encrypted_virtual_filesystem_push_custom_chunking<
761        T: ObjectSource,
762        P: Into<PathBuf> + Send,
763    >(
764        &self,
765        source: T,
766        virtual_directory: P,
767        chunk_size: usize,
768        security_level: SecurityLevel,
769    ) -> Result<(), NetworkError> {
770        self.can_use_revfs()?;
771        let mut virtual_path = virtual_directory.into();
772        virtual_path = prepare_virtual_path(virtual_path);
773        validate_virtual_path(&virtual_path).map_err(|err| {
774            citadel_io::error!(
775                citadel_io::ErrorCode::RemoteRevfsInvalidVirtualPath,
776                err.into_string()
777            )
778        })?;
779        let tx_type = TransferType::RemoteEncryptedVirtualFilesystem {
780            virtual_path,
781            security_level,
782        };
783        self.send_file_with_custom_opts(source, chunk_size, tx_type)
784            .await
785    }
786
787    /// Sends a file to the provided target using the default chunking size with local encryption.
788    /// Only this local node may decrypt the information send to the adjacent node.
789    async fn remote_encrypted_virtual_filesystem_push<T: ObjectSource, P: Into<PathBuf> + Send>(
790        &self,
791        source: T,
792        virtual_directory: P,
793        security_level: SecurityLevel,
794    ) -> Result<(), NetworkError> {
795        self.remote_encrypted_virtual_filesystem_push_custom_chunking(
796            source,
797            virtual_directory,
798            0,
799            security_level,
800        )
801        .await
802    }
803
804    /// Pulls a virtual file from the RE-VFS. If `delete_on_pull` is true, then, the virtual file
805    /// will be taken from the RE-VFS
806    async fn remote_encrypted_virtual_filesystem_pull<P: Into<PathBuf> + Send>(
807        &self,
808        virtual_directory: P,
809        transfer_security_level: SecurityLevel,
810        delete_on_pull: bool,
811    ) -> Result<PathBuf, NetworkError> {
812        self.can_use_revfs()?;
813        let request = NodeRequest::PullObject(PullObject {
814            v_conn: *self.user(),
815            virtual_dir: virtual_directory.into(),
816            delete_on_pull,
817            transfer_security_level,
818        });
819
820        let mut stream = self.remote().send_callback_subscription(request).await?;
821
822        while let Some(event) = stream.next().await {
823            match event.into_result()? {
824                NodeResult::ObjectTransferHandle(ObjectTransferHandle { mut handle, .. }) => {
825                    return handle.receive_file().await.map_err(|err| {
826                        citadel_io::error!(
827                            citadel_io::ErrorCode::RemoteFileTransferFailed,
828                            err.into_string()
829                        )
830                    });
831                }
832
833                NodeResult::PeerEvent(PeerEvent {
834                    event: PeerSignal::SignalReceived { .. },
835                    ..
836                }) => {}
837
838                // The server's refusal (no such file, or it cannot send it) carries its reason.
839                NodeResult::ReVFS(ReVFSResult {
840                    error_message: Some(error),
841                    ..
842                }) => {
843                    return Err(citadel_io::error!(
844                        citadel_io::ErrorCode::RemoteFileTransferFailed,
845                        error
846                    ));
847                }
848
849                NodeResult::ReVFS(ReVFSResult {
850                    error_message: None,
851                    ..
852                }) => {}
853
854                res => {
855                    log::error!(target: "citadel", "Invalid NodeResult for REVFS FileTransfer request received: {res:?}");
856                    return Err(citadel_io::error!(
857                        citadel_io::ErrorCode::RemoteRevfsInvalidResponse
858                    ));
859                }
860            }
861        }
862
863        Err(citadel_io::error!(
864            citadel_io::ErrorCode::RemoteRevfsFileTransferStreamDied
865        ))
866    }
867
868    /// Deletes the file from the RE-VFS. If the contents are desired on delete,
869    /// consider calling `Self::remote_encrypted_virtual_filesystem_pull` with the delete
870    /// parameter set to true
871    async fn remote_encrypted_virtual_filesystem_delete<P: Into<PathBuf> + Send>(
872        &self,
873        virtual_directory: P,
874    ) -> Result<(), NetworkError> {
875        self.can_use_revfs()?;
876        let request = NodeRequest::DeleteObject(DeleteObject {
877            v_conn: *self.user(),
878            virtual_dir: virtual_directory.into(),
879            security_level: Default::default(),
880        });
881
882        let mut stream = self.remote().send_callback_subscription(request).await?;
883        while let Some(event) = stream.next().await {
884            match event.into_result()? {
885                NodeResult::ReVFS(result) => {
886                    return if let Some(error) = result.error_message {
887                        Err(citadel_io::error!(
888                            citadel_io::ErrorCode::RemoteFileTransferFailed,
889                            error
890                        ))
891                    } else {
892                        Ok(())
893                    }
894                }
895
896                evt => {
897                    log::error!(target: "citadel", "Invalid NodeResult for REVFS Delete request received: {evt:?}");
898                }
899            }
900        }
901
902        Err(citadel_io::error!(
903            citadel_io::ErrorCode::RemoteRevfsDeleteStreamDied
904        ))
905    }
906
907    /// Connects to the peer with custom settings
908    async fn connect_to_peer_custom(
909        &self,
910        session_security_settings: SessionSecuritySettings,
911        udp_mode: UdpMode,
912        peer_session_password: Option<PreSharedKey>,
913    ) -> Result<PeerConnectSuccess<R>, NetworkError> {
914        use std::time::Duration;
915
916        // Timeout for the entire P2P connection process.
917        // This prevents indefinite hangs when the server never responds with PeerChannelCreated.
918        // The timeout should be long enough for normal hole punching (which has its own 30s timeout)
919        // plus key exchange, but short enough to fail fast when something is stuck.
920        const P2P_CONNECT_TIMEOUT: Duration = Duration::from_secs(60);
921
922        let session_cid = self.user().get_session_cid();
923        let peer_target = self.try_as_peer_connection().await?;
924
925        let mut stream = self
926            .remote()
927            .send_callback_subscription(NodeRequest::PeerCommand(PeerCommand {
928                session_cid,
929                command: PeerSignal::PostConnect {
930                    peer_conn_type: peer_target,
931                    ticket_opt: None,
932                    invitee_response: None,
933                    session_security_settings,
934                    udp_mode,
935                    session_password: peer_session_password,
936                },
937            }))
938            .await?;
939
940        let connect_task = async {
941            while let Some(status) = stream.next().await {
942                match status.into_result()? {
943                    NodeResult::PeerChannelCreated(PeerChannelCreated {
944                        ticket: _,
945                        channel,
946                        udp_rx_opt,
947                    }) => {
948                        let username = self.target_username().map(ToString::to_string);
949                        let remote = PeerRemote {
950                            inner: self.remote().clone(),
951                            peer: peer_target.as_virtual_connection(),
952                            username,
953                            session_security_settings,
954                        };
955
956                        return Ok(PeerConnectSuccess {
957                            remote,
958                            channel: *channel,
959                            udp_channel_rx: udp_rx_opt,
960                            incoming_object_transfer_handles: None,
961                        });
962                    }
963
964                    NodeResult::PeerEvent(PeerEvent {
965                        event:
966                            PeerSignal::PostConnect {
967                                invitee_response, ..
968                            },
969                        ..
970                    }) => match invitee_response {
971                        Some(PeerResponse::Timeout) => {
972                            return Err(citadel_io::error!(
973                                citadel_io::ErrorCode::RemotePeerNoResponse
974                            ))
975                        }
976                        Some(PeerResponse::Decline) => {
977                            return Err(citadel_io::error!(
978                                citadel_io::ErrorCode::RemotePeerDeclined
979                            ))
980                        }
981                        _ => {}
982                    },
983
984                    // A routing failure for this ticket (e.g. the server could not deliver our
985                    // accept or request). Fail fast with the reason instead of waiting out the
986                    // full 60s watchdog — the caller can retry immediately.
987                    NodeResult::PeerEvent(PeerEvent {
988                        event: PeerSignal::SignalError { error, .. },
989                        ..
990                    }) => {
991                        return Err(citadel_io::error!(
992                            citadel_io::ErrorCode::RemoteP2pSignalRoutingFailed,
993                            error
994                        ))
995                    }
996
997                    _ => {}
998                }
999            }
1000
1001            Err(citadel_io::error!(
1002                citadel_io::ErrorCode::RemoteKernelStreamDied,
1003                "connect_to_peer_custom"
1004            ))
1005        };
1006
1007        match citadel_io::time::timeout(P2P_CONNECT_TIMEOUT, connect_task).await {
1008            Ok(result) => result,
1009            Err(_elapsed) => Err(citadel_io::error!(
1010                citadel_io::ErrorCode::RemoteP2pConnectTimeout,
1011                P2P_CONNECT_TIMEOUT.as_secs()
1012            )),
1013        }
1014    }
1015
1016    /// Supplies (`Some`) or withdraws (`None`) the TURN relay configuration for the next P2P
1017    /// attempt with this peer. Both peers must supply one for the relay to be used (the
1018    /// accepting peer before it accepts); the attempt consumes it. When direct P2P is impossible
1019    /// or fails, the peers then connect through the relay, and
1020    /// [`PeerChannel::p2p_path`](citadel_proto::prelude::PeerChannel::p2p_path) reports
1021    /// [`P2pPath::Turn`].
1022    async fn set_turn_config(
1023        &self,
1024        config: Option<citadel_proto::prelude::TurnRelayConfig>,
1025    ) -> Result<(), NetworkError> {
1026        let session_cid = self.user().get_session_cid();
1027        let peer_cid = self
1028            .try_as_peer_connection()
1029            .await?
1030            .get_original_target_cid();
1031        self.remote()
1032            .send(NodeRequest::SetPeerTurnConfig(SetPeerTurnConfig {
1033                session_cid,
1034                peer_cid,
1035                config,
1036            }))
1037            .await
1038            .map(|_| ())
1039    }
1040
1041    /// [`Self::connect_to_peer_custom`] with a TURN relay configuration for this attempt.
1042    async fn connect_to_peer_with_turn(
1043        &self,
1044        session_security_settings: SessionSecuritySettings,
1045        udp_mode: UdpMode,
1046        peer_session_password: Option<PreSharedKey>,
1047        turn: citadel_proto::prelude::TurnRelayConfig,
1048    ) -> Result<PeerConnectSuccess<R>, NetworkError> {
1049        self.set_turn_config(Some(turn)).await?;
1050        self.connect_to_peer_custom(session_security_settings, udp_mode, peer_session_password)
1051            .await
1052    }
1053
1054    /// Connects to the target peer with default settings
1055    async fn connect_to_peer(&self) -> Result<PeerConnectSuccess<R>, NetworkError> {
1056        self.connect_to_peer_custom(Default::default(), Default::default(), Default::default())
1057            .await
1058    }
1059
1060    /// Posts a registration request to a peer
1061    async fn register_to_peer(&self) -> Result<PeerRegisterStatus, NetworkError> {
1062        let session_cid = self.user().get_session_cid();
1063        let peer_target = self.try_as_peer_connection().await?;
1064        // TODO: Get rid of this step. Should be handled by the protocol
1065        let local_username = self
1066            .remote()
1067            .account_manager()
1068            .get_username_by_cid(session_cid)
1069            .await?
1070            .ok_or_else(|| citadel_io::error!(citadel_io::ErrorCode::RemoteLocalUsernameMissing))?;
1071        let peer_username_opt = self.target_username().map(ToString::to_string);
1072
1073        let mut stream = self
1074            .remote()
1075            .send_callback_subscription(NodeRequest::PeerCommand(PeerCommand {
1076                session_cid,
1077                command: PeerSignal::PostRegister {
1078                    peer_conn_type: peer_target,
1079                    inviter_username: local_username,
1080                    invitee_username: peer_username_opt,
1081                    ticket_opt: None,
1082                    invitee_response: None,
1083                },
1084            }))
1085            .await?;
1086
1087        while let Some(status) = stream.next().await {
1088            if let NodeResult::PeerEvent(PeerEvent {
1089                event:
1090                    PeerSignal::PostRegister {
1091                        peer_conn_type: _,
1092                        inviter_username: _,
1093                        invitee_username: _,
1094                        ticket_opt: _,
1095                        invitee_response: Some(resp),
1096                    },
1097                ..
1098            }) = status.into_result()?
1099            {
1100                match resp {
1101                    PeerResponse::Accept(..) => return Ok(PeerRegisterStatus::Accepted),
1102                    PeerResponse::Decline => return Ok(PeerRegisterStatus::Declined),
1103                    PeerResponse::Timeout => return Ok(PeerRegisterStatus::Failed { reason: Some("Timeout on register request. Peer did not accept in time. Try again later".to_string()) }),
1104                    _ => {}
1105                }
1106            }
1107        }
1108
1109        Err(citadel_io::error!(
1110            citadel_io::ErrorCode::RemoteKernelStreamDied,
1111            format!("register_to_peer: {:?}", stream.callback_key())
1112        ))
1113    }
1114
1115    /// Deregisters the currently locked target. If the target is a client to server
1116    /// connection, deregisters from the server. If the target is a p2p connection,
1117    /// deregisters the p2p
1118    async fn deregister(&self) -> Result<(), NetworkError> {
1119        if let Ok(peer_conn) = self.try_as_peer_connection().await {
1120            let peer_request = PeerSignal::Deregister {
1121                peer_conn_type: peer_conn,
1122            };
1123            let session_cid = self.user().get_session_cid();
1124            let request = NodeRequest::PeerCommand(PeerCommand {
1125                session_cid,
1126                command: peer_request,
1127            });
1128
1129            let mut subscription = self.remote().send_callback_subscription(request).await?;
1130            while let Some(result) = subscription.next().await {
1131                if let NodeResult::PeerEvent(PeerEvent {
1132                    event: PeerSignal::DeregistrationSuccess { .. },
1133                    ..
1134                }) = result.into_result()?
1135                {
1136                    return Ok(());
1137                }
1138            }
1139        } else {
1140            // c2s conn
1141            let cid = self.user().get_session_cid();
1142            let request = NodeRequest::DeregisterFromHypernode(DeregisterFromHypernode {
1143                session_cid: cid,
1144                v_conn_type: *self.user(),
1145            });
1146            let mut subscription = self.remote().send_callback_subscription(request).await?;
1147            while let Some(result) = subscription.next().await {
1148                match result.into_result()? {
1149                    NodeResult::DeRegistration(DeRegistration {
1150                        session_cid: _,
1151                        ticket_opt: _,
1152                        success: true,
1153                    }) => return Ok(()),
1154                    NodeResult::DeRegistration(DeRegistration {
1155                        session_cid: _,
1156                        ticket_opt: _,
1157                        success: false,
1158                    }) => {
1159                        return Err(citadel_io::error!(
1160                            citadel_io::ErrorCode::RemoteDeregisterFailed
1161                        ))
1162                    }
1163
1164                    _ => {}
1165                }
1166            }
1167        }
1168
1169        Err(citadel_io::error!(
1170            citadel_io::ErrorCode::RemoteDeregisterEndedUnexpectedly
1171        ))
1172    }
1173
1174    async fn disconnect(&self) -> Result<(), NetworkError> {
1175        if let Ok(peer_conn) = self.try_as_peer_connection().await {
1176            if let PeerConnectionType::LocalGroupPeer {
1177                session_cid,
1178                peer_cid: _,
1179            } = peer_conn
1180            {
1181                let request = NodeRequest::PeerCommand(PeerCommand {
1182                    session_cid,
1183                    command: PeerSignal::Disconnect {
1184                        peer_conn_type: peer_conn,
1185                        disconnect_response: None,
1186                        disconnect_token: None,
1187                    },
1188                });
1189
1190                let mut subscription = self.remote().send_callback_subscription(request).await?;
1191
1192                while let Some(event) = subscription.next().await {
1193                    if let NodeResult::PeerEvent(PeerEvent {
1194                        event:
1195                            PeerSignal::Disconnect {
1196                                peer_conn_type: _,
1197                                disconnect_response: Some(_),
1198                                ..
1199                            },
1200                        ..
1201                    }) = event.into_result()?
1202                    {
1203                        return Ok(());
1204                    }
1205                }
1206
1207                Err(citadel_io::error!(
1208                    citadel_io::ErrorCode::RemoteDisconnectEventMissing
1209                ))
1210            } else {
1211                Err(citadel_io::error!(
1212                    citadel_io::ErrorCode::RemoteExternalGroupPeerUnsupported
1213                ))
1214            }
1215        } else {
1216            //c2s conn
1217            let cid = self.user().get_session_cid();
1218            let request =
1219                NodeRequest::DisconnectFromHypernode(DisconnectFromHypernode { session_cid: cid });
1220
1221            let mut subscription = self.remote().send_callback_subscription(request).await?;
1222            // Bounded, because this wait is otherwise the end of the line.
1223            //
1224            // The stream closing is handled below, but a subscription that
1225            // stays OPEN and simply never carries the matching Disconnect
1226            // parks here for ever. That is the reconnection suite's 240s
1227            // timeout: with the phase markers reaching CI, both peers finish
1228            // phase one and the disconnecting side never logs the line after
1229            // this call, while the other blocks on the barrier it never
1230            // reaches. The only await between those two markers is this one.
1231            //
1232            // The session is being torn down either way, so an unconfirmed
1233            // disconnect is reported rather than waited on: a caller can retry
1234            // or proceed, and neither is possible from inside an indefinite
1235            // await.
1236            const DISCONNECT_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(30);
1237            let deadline = citadel_io::time::Instant::now() + DISCONNECT_CONFIRMATION_TIMEOUT;
1238            // What the subscription DID carry, for the occasion when it does not
1239            // carry a Disconnect.
1240            //
1241            // `RemoteDisconnectEventMissing` says only that nothing matched in
1242            // thirty seconds, which cannot distinguish the two things it might
1243            // mean. A NON-ZERO count says the subscription was alive and other
1244            // events reached it, so the Disconnect was never emitted or was
1245            // emitted under a key this subscription does not hold. ZERO says the
1246            // subscription heard nothing at all, which points at the session
1247            // rather than at the routing. Those want different fixes, and the
1248            // error tells them apart not at all.
1249            //
1250            // This is the same instrument that localised the UDP one-shot race:
1251            // an intermittent CI failure that no local run reproduces is only
1252            // ever going to be solved by what it says about itself when it next
1253            // happens.
1254            let mut carried = 0usize;
1255            loop {
1256                let remaining =
1257                    deadline.saturating_duration_since(citadel_io::time::Instant::now());
1258                if remaining.is_zero() {
1259                    log::warn!(target: "citadel", "[dc-wait] no Disconnect for cid {cid} within {DISCONNECT_CONFIRMATION_TIMEOUT:?}; the subscription carried {carried} other event(s)");
1260                    return Err(citadel_io::error!(
1261                        citadel_io::ErrorCode::RemoteDisconnectEventMissing
1262                    ));
1263                }
1264                match citadel_io::time::timeout(remaining, subscription.next()).await {
1265                    Ok(Some(event)) => {
1266                        carried += 1;
1267                        if let NodeResult::Disconnect(Disconnect {
1268                            success, message, ..
1269                        }) = event.into_result()?
1270                        {
1271                            return if success {
1272                                Ok(())
1273                            } else {
1274                                Err(citadel_io::error!(
1275                                    citadel_io::ErrorCode::RemoteDisconnected,
1276                                    message
1277                                ))
1278                            };
1279                        }
1280                    }
1281                    // The stream ended without ever carrying the event.
1282                    Ok(None) => {
1283                        log::warn!(target: "citadel", "[dc-wait] the subscription for cid {cid} closed without a Disconnect; it carried {carried} other event(s)");
1284                        break;
1285                    }
1286                    Err(_elapsed) => {
1287                        log::warn!(target: "citadel", "[dc-wait] timed out waiting for a Disconnect for cid {cid}; the subscription carried {carried} other event(s)");
1288                        return Err(citadel_io::error!(
1289                            citadel_io::ErrorCode::RemoteDisconnectEventMissing
1290                        ));
1291                    }
1292                }
1293            }
1294
1295            Err(citadel_io::error!(
1296                citadel_io::ErrorCode::RemoteDisconnectEventMissing
1297            ))
1298        }
1299    }
1300
1301    async fn create_group(
1302        &self,
1303        initial_users_to_invite: Option<Vec<UserIdentifier>>,
1304    ) -> Result<GroupChannel, NetworkError> {
1305        self.create_group_with_options(initial_users_to_invite, MessageGroupOptions::default())
1306            .await
1307    }
1308
1309    /// Create a group with explicit [`MessageGroupOptions`] — e.g. a zero-trust
1310    /// [`GroupHierarchyMode::CommandHierarchy`](citadel_types::proto::GroupHierarchyMode) where a
1311    /// superior can read its subordinates' messages. `options.hierarchy` carries the initial rank
1312    /// assignment (member cid → command path); it is consumed owner-locally and never reaches the relay.
1313    async fn create_group_with_options(
1314        &self,
1315        initial_users_to_invite: Option<Vec<UserIdentifier>>,
1316        options: MessageGroupOptions,
1317    ) -> Result<GroupChannel, NetworkError> {
1318        let session_cid = self.user().get_session_cid();
1319
1320        let mut initial_users = vec![];
1321        // NOTE: default is PRIVATE mode, meaning all users in group must be registered to the owner.
1322        // Initial users are UserIdentifiers resolved to cids below.
1323        if let Some(initial_users_to_invite) = initial_users_to_invite {
1324            for user in initial_users_to_invite {
1325                initial_users.push(
1326                    self.remote()
1327                        .account_manager()
1328                        .find_target_information(session_cid, user.clone())
1329                        .await?
1330                        .ok_or_else(|| {
1331                            citadel_io::error!(
1332                                citadel_io::ErrorCode::RemoteGroupAccountNotFound,
1333                                citadel_io::Dbg(user),
1334                                citadel_io::Dbg(session_cid)
1335                            )
1336                        })
1337                        .map(|r| r.1.cid)?,
1338                )
1339            }
1340        }
1341
1342        let group_request = GroupBroadcast::Create {
1343            initial_invitees: initial_users,
1344            options,
1345        };
1346        let request = NodeRequest::GroupBroadcastCommand(GroupBroadcastCommand {
1347            session_cid,
1348            command: group_request,
1349        });
1350        let mut subscription = self.remote().send_callback_subscription(request).await?;
1351
1352        // The loop lives in `group_create_wait` so its refusal paths can be
1353        // tested without a running node: the only way to make a real server
1354        // refuse a Create is to give the owner 256 groups first.
1355        //
1356        // It matched GroupChannelCreated and nothing else, and — alone among its
1357        // neighbours — never called into_result(). So SignalError,
1358        // OutboundRequestRejected and InternalServerError were discarded, and so
1359        // was the server's own answer to a Create it could not perform:
1360        // CreateResponse { key: None }, delivered on this very ticket. Every one
1361        // of those left the caller parked on a stream that would never speak
1362        // again. The broadcast prefab handles that same event as
1363        // BroadcastCreateGroupFailed — the guard existed, in the prefab only.
1364        match crate::group_create_wait::await_group_creation(&mut subscription).await? {
1365            crate::group_create_wait::GroupCreation::Created(channel) => Ok(channel),
1366            crate::group_create_wait::GroupCreation::Refused => Err(citadel_io::error!(
1367                citadel_io::ErrorCode::Generic,
1368                "The server refused to create the group"
1369            )),
1370            crate::group_create_wait::GroupCreation::Ended => Err(citadel_io::error!(
1371                citadel_io::ErrorCode::RemoteCreateGroupEndedUnexpectedly
1372            )),
1373        }
1374    }
1375
1376    /// Lists all groups that which the current peer owns
1377    async fn list_owned_groups(&self) -> Result<Vec<MessageGroupKey>, NetworkError> {
1378        let session_cid = self.user().get_session_cid();
1379        let cid_to_check_for = match self.try_as_peer_connection().await {
1380            Ok(res) => res.get_original_target_cid(),
1381            _ => session_cid,
1382        };
1383        let group_request = GroupBroadcast::ListGroupsFor {
1384            cid: cid_to_check_for,
1385        };
1386        let request = NodeRequest::GroupBroadcastCommand(GroupBroadcastCommand {
1387            session_cid,
1388            command: group_request,
1389        });
1390
1391        let mut subscription = self.remote().send_callback_subscription(request).await?;
1392
1393        while let Some(evt) = subscription.next().await {
1394            if let NodeResult::GroupEvent(GroupEvent {
1395                session_cid: _,
1396                ticket: _,
1397                event: GroupBroadcast::ListResponse { groups },
1398            }) = evt.into_result()?
1399            {
1400                return Ok(groups);
1401            }
1402        }
1403
1404        Err(citadel_io::error!(
1405            citadel_io::ErrorCode::RemoteListGroupsEndedUnexpectedly
1406        ))
1407    }
1408
1409    /// Lists all active sessions, including the local nat type. For each active session,
1410    /// lists each connection info which includes the remote nat type, connection status,
1411    /// peer id, and latest ratchet version.
1412    async fn list_sessions(&self) -> Result<ActiveSessions, NetworkError> {
1413        let request = NodeRequest::GetActiveSessions;
1414        let mut subscription = self.remote().send_callback_subscription(request).await?;
1415
1416        if let Some(NodeResult::SessionList(result)) = subscription.next().await {
1417            return Ok(result.sessions);
1418        }
1419
1420        Err(citadel_io::error!(
1421            citadel_io::ErrorCode::RemoteListSessionsEndedUnexpectedly
1422        ))
1423    }
1424
1425    /// Begins a re-key, updating the container in the process.
1426    /// Returns the new key matrix version. Does not return the new key version
1427    /// if the rekey fails, or, if a current rekey is already executing
1428    async fn rekey(&self) -> Result<Option<u32>, NetworkError> {
1429        let request = NodeRequest::ReKey(ReKey {
1430            v_conn_type: *self.user(),
1431        });
1432        let mut subscription = self.remote().send_callback_subscription(request).await?;
1433
1434        while let Some(evt) = subscription.next().await {
1435            if let NodeResult::ReKeyResult(result) = evt {
1436                return match result.status {
1437                    ReKeyReturnType::Success { version } => Ok(Some(version)),
1438                    ReKeyReturnType::AlreadyInProgress => Ok(None),
1439                    ReKeyReturnType::Failure { err } => Err(citadel_io::error!(
1440                        citadel_io::ErrorCode::RemoteRekeyFailed,
1441                        err
1442                    )),
1443                };
1444            }
1445        }
1446
1447        Err(citadel_io::error!(
1448            citadel_io::ErrorCode::RemoteRekeyEndedUnexpectedly
1449        ))
1450    }
1451
1452    /// Checks if the locked target is registered
1453    async fn is_peer_registered(&self) -> Result<bool, NetworkError> {
1454        let target = self.try_as_peer_connection().await?;
1455        if let PeerConnectionType::LocalGroupPeer {
1456            session_cid: local_cid,
1457            peer_cid,
1458        } = target
1459        {
1460            let peers = self.remote().get_local_group_peers(local_cid, None).await?;
1461            citadel_logging::info!(target: "citadel", "Checking to see if {target} is registered in {peers:?}");
1462            Ok(peers.iter().any(|p| p.cid == peer_cid))
1463        } else {
1464            Err(citadel_io::error!(
1465                citadel_io::ErrorCode::RemoteExternalGroupPeerUnsupportedYet
1466            ))
1467        }
1468    }
1469
1470    #[doc(hidden)]
1471    async fn try_as_peer_connection(&self) -> Result<PeerConnectionType, NetworkError> {
1472        let verified_return = |user: &VirtualTargetType| {
1473            user.try_as_peer_connection().ok_or(citadel_io::error!(
1474                citadel_io::ErrorCode::RemoteTargetNotPeer
1475            ))
1476        };
1477
1478        if self.user().get_target_cid() == 0 {
1479            // in this case, the user re-used a remote locked to a registration target
1480            // where the username was provided, but the cid was 0 (unknown).
1481            let peer_username = self.target_username().ok_or_else(|| {
1482                citadel_io::error!(citadel_io::ErrorCode::RemoteTargetCidZeroNoUsername)
1483            })?;
1484            let session_cid = self.user().get_session_cid();
1485            let expected_peer_cid = self
1486                .remote()
1487                .account_manager()
1488                .get_persistence_handler()
1489                .get_cid_by_username(peer_username);
1490            // get the peer cid from the account manager (implying the peers are already registered).
1491            // fallback to the mapped cid if the peer is not registered
1492            let peer_cid = self
1493                .remote()
1494                .account_manager()
1495                .find_target_information(session_cid, peer_username)
1496                .await?
1497                .map(|r| r.1.cid)
1498                .unwrap_or(expected_peer_cid);
1499
1500            let mut user = *self.user();
1501            user.set_target_cid(peer_cid);
1502            verified_return(&user)
1503        } else {
1504            verified_return(self.user())
1505        }
1506    }
1507
1508    #[doc(hidden)]
1509    fn can_use_revfs(&self) -> Result<(), NetworkError> {
1510        if let Some(sess) = self.session_security_settings() {
1511            if sess.crypto_params.kem_algorithm == KemAlgorithm::MlKem {
1512                Ok(())
1513            } else {
1514                Err(citadel_io::error!(
1515                    citadel_io::ErrorCode::RemoteRevfsRequiresKyber
1516                ))
1517            }
1518        } else {
1519            Err(citadel_io::error!(
1520                citadel_io::ErrorCode::RemoteRevfsUnsupportedRemote
1521            ))
1522        }
1523    }
1524}
1525
1526impl<T: TargetLockedRemote<R>, R: Ratchet> ProtocolRemoteTargetExt<R> for T {}
1527
1528pub mod results {
1529    use crate::prefabs::client::peer_connection::FileTransferHandleRx;
1530    use crate::prelude::{PeerChannel, UdpChannel};
1531    use crate::remote_ext::remote_specialization::PeerRemote;
1532    use citadel_io::tokio::sync::oneshot::Receiver;
1533    use citadel_proto::prelude::*;
1534    use std::fmt::Debug;
1535
1536    pub struct PeerConnectSuccess<R: Ratchet> {
1537        pub channel: PeerChannel<R>,
1538        pub udp_channel_rx: Option<Receiver<UdpChannel<R>>>,
1539        pub remote: PeerRemote<R>,
1540        /// Receives incoming file/object transfer requests. The handles must be
1541        /// .accepted() before the file/object transfer is allowed to proceed
1542        pub(crate) incoming_object_transfer_handles: Option<FileTransferHandleRx>,
1543    }
1544
1545    impl<R: Ratchet> Debug for PeerConnectSuccess<R> {
1546        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1547            f.debug_struct("PeerConnectSuccess")
1548                .field("channel", &self.channel)
1549                .field("udp_channel_rx", &self.udp_channel_rx)
1550                .finish()
1551        }
1552    }
1553
1554    impl<R: Ratchet> PeerConnectSuccess<R> {
1555        /// Obtains a receiver which yields incoming file/object transfer handles
1556        pub fn get_incoming_file_transfer_handle(
1557            &mut self,
1558        ) -> Result<FileTransferHandleRx, NetworkError> {
1559            self.incoming_object_transfer_handles
1560                .take()
1561                .ok_or(citadel_io::error!(
1562                    citadel_io::ErrorCode::RemoteFunctionAlreadyCalled
1563                ))
1564        }
1565    }
1566
1567    /// The peer's answer to a registration request.
1568    ///
1569    /// `Ok` means the request completed, not that it was accepted — a decline is
1570    /// a successful round trip with the answer "no". This type derived nothing,
1571    /// so a caller could neither compare it nor log it, and all three real
1572    /// callers wrote `Ok(_)` / `let _ =` and carried on as though the peer had
1573    /// said yes. `peer_connection.rs` then issued a PostConnect to a peer that
1574    /// had just declined and waited out a 60s RemoteP2pConnectTimeout, reporting
1575    /// that instead of the decline the SDK had in hand.
1576    ///
1577    /// The derives are the fix for the type; `is_accepted` is the fix for the
1578    /// call sites, which needed something shorter to write than the mistake.
1579    #[derive(Clone, Debug, PartialEq, Eq)]
1580    pub enum PeerRegisterStatus {
1581        Accepted,
1582        Declined,
1583        Failed { reason: Option<String> },
1584    }
1585
1586    impl PeerRegisterStatus {
1587        /// Did the peer say yes?
1588        pub fn is_accepted(&self) -> bool {
1589            matches!(self, PeerRegisterStatus::Accepted)
1590        }
1591
1592        /// The reason this was not an acceptance, or `None` if it was.
1593        pub fn refusal_reason(&self) -> Option<String> {
1594            match self {
1595                PeerRegisterStatus::Accepted => None,
1596                PeerRegisterStatus::Declined => Some("The peer declined the request".to_string()),
1597                PeerRegisterStatus::Failed { reason } => Some(
1598                    reason
1599                        .clone()
1600                        .unwrap_or_else(|| "The registration request failed".to_string()),
1601                ),
1602            }
1603        }
1604    }
1605
1606    #[derive(Clone, Debug)]
1607    pub struct LocalGroupPeer {
1608        pub cid: u64,
1609        pub is_online: bool,
1610    }
1611
1612    #[derive(Clone, Debug)]
1613    pub struct LocalGroupPeerFullInfo {
1614        pub cid: u64,
1615        pub username: Option<String>,
1616        pub full_name: Option<String>,
1617        pub is_online: bool,
1618    }
1619}
1620
1621pub mod remote_specialization {
1622    use crate::prelude::*;
1623    use std::ops::{Deref, DerefMut};
1624
1625    #[derive(Debug, Clone)]
1626    pub struct PeerRemote<R: Ratchet> {
1627        pub(crate) inner: NodeRemote<R>,
1628        pub(crate) peer: VirtualTargetType,
1629        pub(crate) username: Option<String>,
1630        pub(crate) session_security_settings: SessionSecuritySettings,
1631    }
1632
1633    impl<R: Ratchet> Deref for PeerRemote<R> {
1634        type Target = NodeRemote<R>;
1635        fn deref(&self) -> &Self::Target {
1636            &self.inner
1637        }
1638    }
1639
1640    impl<R: Ratchet> DerefMut for PeerRemote<R> {
1641        fn deref_mut(&mut self) -> &mut Self::Target {
1642            &mut self.inner
1643        }
1644    }
1645
1646    impl<R: Ratchet> TargetLockedRemote<R> for PeerRemote<R> {
1647        fn user(&self) -> &VirtualTargetType {
1648            &self.peer
1649        }
1650        fn remote(&self) -> &NodeRemote<R> {
1651            &self.inner
1652        }
1653        fn target_username(&self) -> Option<&str> {
1654            self.username.as_deref()
1655        }
1656        fn user_mut(&mut self) -> &mut VirtualTargetType {
1657            &mut self.peer
1658        }
1659
1660        fn session_security_settings(&self) -> Option<&SessionSecuritySettings> {
1661            Some(&self.session_security_settings)
1662        }
1663    }
1664}
1665
1666#[cfg(all(test, not(target_family = "wasm")))]
1667mod tests {
1668    use crate::prefabs::client::single_connection::SingleClientServerConnectionKernel;
1669    use crate::prefabs::client::DefaultServerConnectionSettingsBuilder;
1670    use crate::prelude::*;
1671    use citadel_io::tokio;
1672    use rstest::rstest;
1673    use std::net::SocketAddr;
1674    use std::sync::atomic::{AtomicBool, Ordering};
1675    use std::sync::Arc;
1676    use uuid::Uuid;
1677
1678    pub struct ReceiverFileTransferKernel<R: Ratchet>(
1679        pub Option<NodeRemote<R>>,
1680        pub Arc<AtomicBool>,
1681    );
1682
1683    #[async_trait]
1684    impl<R: Ratchet> NetKernel<R> for ReceiverFileTransferKernel<R> {
1685        fn load_remote(&mut self, node_remote: NodeRemote<R>) -> Result<(), NetworkError> {
1686            self.0 = Some(node_remote);
1687            Ok(())
1688        }
1689
1690        async fn on_start(&self) -> Result<(), NetworkError> {
1691            Ok(())
1692        }
1693
1694        async fn on_node_event_received(&self, message: NodeResult<R>) -> Result<(), NetworkError> {
1695            log::trace!(target: "citadel", "SERVER received {:?}", message);
1696            if let NodeResult::ObjectTransferHandle(ObjectTransferHandle { mut handle, .. }) =
1697                message.into_result()?
1698            {
1699                let mut path = None;
1700                // accept the transfer
1701                handle
1702                    .accept()
1703                    .map_err(|err| NetworkError::msg(err.into_string()))?;
1704
1705                use citadel_types::proto::ObjectTransferStatus;
1706                use futures::StreamExt;
1707                while let Some(status) = handle.next().await {
1708                    match status {
1709                        ObjectTransferStatus::ReceptionComplete => {
1710                            log::trace!(target: "citadel", "Server has finished receiving the file!");
1711                            let cmp = include_bytes!("../../resources/TheBridge.pdf");
1712                            let streamed_data = citadel_io::tokio::fs::read(path.clone().unwrap())
1713                                .await
1714                                .unwrap();
1715                            assert_eq!(
1716                                cmp,
1717                                streamed_data.as_slice(),
1718                                "Original data and streamed data does not match"
1719                            );
1720
1721                            self.1.store(true, Ordering::Relaxed);
1722                            self.0.clone().unwrap().shutdown().await?;
1723                        }
1724
1725                        ObjectTransferStatus::ReceptionBeginning(file_path, vfm) => {
1726                            path = Some(file_path);
1727                            assert_eq!(vfm.name, "TheBridge.pdf")
1728                        }
1729
1730                        _ => {}
1731                    }
1732                }
1733            }
1734
1735            Ok(())
1736        }
1737
1738        async fn on_stop(&mut self) -> Result<(), NetworkError> {
1739            Ok(())
1740        }
1741    }
1742
1743    pub fn server_info<'a, R: Ratchet>(
1744        switch: Arc<AtomicBool>,
1745    ) -> (NodeFuture<'a, ReceiverFileTransferKernel<R>>, SocketAddr) {
1746        crate::test_common::server_test_node(ReceiverFileTransferKernel(None, switch), |_| {})
1747    }
1748
1749    #[rstest]
1750    #[case(
1751        EncryptionAlgorithm::AES_GCM_256,
1752        KemAlgorithm::MlKem,
1753        SigAlgorithm::None
1754    )]
1755    #[case(
1756        EncryptionAlgorithm::MlKemHybrid,
1757        KemAlgorithm::MlKem,
1758        SigAlgorithm::MlDsa65
1759    )]
1760    #[timeout(std::time::Duration::from_secs(90))]
1761    #[tokio::test]
1762    async fn test_c2s_file_transfer(
1763        #[case] enx: EncryptionAlgorithm,
1764        #[case] kem: KemAlgorithm,
1765        #[case] sig: SigAlgorithm,
1766    ) {
1767        citadel_logging::setup_log();
1768        let client_success = &AtomicBool::new(false);
1769        let server_success = &Arc::new(AtomicBool::new(false));
1770        let (server, server_addr) = server_info::<StackedRatchet>(server_success.clone());
1771        let uuid = Uuid::new_v4();
1772
1773        let session_security_settings = SessionSecuritySettingsBuilder::default()
1774            .with_crypto_params(enx + kem + sig)
1775            .build()
1776            .unwrap();
1777
1778        let server_connection_settings =
1779            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
1780                .with_session_security_settings(session_security_settings)
1781                .disable_udp()
1782                .build()
1783                .unwrap();
1784
1785        let client_kernel = SingleClientServerConnectionKernel::new(
1786            server_connection_settings,
1787            |connection| async move {
1788                log::trace!(target: "citadel", "***CLIENT LOGIN SUCCESS :: File transfer next ***");
1789                connection
1790                    .send_file_with_custom_opts(
1791                        "../resources/TheBridge.pdf",
1792                        32 * 1024,
1793                        TransferType::FileTransfer,
1794                    )
1795                    .await
1796                    .unwrap();
1797                log::trace!(target: "citadel", "***CLIENT FILE TRANSFER SUCCESS***");
1798                client_success.store(true, Ordering::Relaxed);
1799                connection.shutdown_kernel().await
1800            },
1801        );
1802
1803        let client = DefaultNodeBuilder::default().build(client_kernel).unwrap();
1804
1805        let joined = futures::future::try_join(server, client);
1806
1807        let _ = joined.await.unwrap();
1808
1809        assert!(client_success.load(Ordering::Relaxed));
1810        assert!(server_success.load(Ordering::Relaxed));
1811    }
1812}
citadel_sdk/remote_ext.rs

citadel_sdk/
remote_ext.rs
