1use crate::prefabs::client::peer_connection::FileTransferHandleRx;
58use crate::prefabs::client::ServerConnectionSettings;
59use crate::prefabs::ClientServerRemote;
60use crate::remote_ext::CitadelClientServerConnection;
61use crate::remote_ext::ProtocolRemoteExt;
62use citadel_io::Mutex;
63use citadel_proto::prelude::*;
64use citadel_types::crypto::PreSharedKey;
65use futures::Future;
66use std::marker::PhantomData;
67use std::net::SocketAddr;
68use uuid::Uuid;
69
70pub struct SingleClientServerConnectionKernel<F, Fut, R: Ratchet> {
74 handler: Mutex<Option<F>>,
75 udp_mode: UdpMode,
76 auth_info: Mutex<Option<ConnectionType>>,
77 session_security_settings: SessionSecuritySettings,
78 unprocessed_signal_filter_tx:
79 Mutex<Option<citadel_io::tokio::sync::mpsc::UnboundedSender<NodeResult<R>>>>,
80 remote: Option<NodeRemote<R>>,
81 server_password: Option<PreSharedKey>,
82 rx_incoming_object_transfer_handle: Mutex<Option<FileTransferHandleRx>>,
83 tx_incoming_object_transfer_handle:
84 citadel_io::tokio::sync::mpsc::UnboundedSender<ObjectTransferHandler>,
85 _pd: PhantomData<fn() -> Fut>,
87}
88
89#[derive(Debug)]
90pub(crate) enum ConnectionType {
91 Register {
92 server_addr: SocketAddr,
93 username: String,
94 password: SecBuffer,
95 full_name: String,
96 },
97 Connect {
98 username: String,
99 password: SecBuffer,
100 },
101 Transient {
102 uuid: Uuid,
103 server_addr: SocketAddr,
104 },
105}
106
107impl<F, Fut, R: Ratchet> SingleClientServerConnectionKernel<F, Fut, R>
108where
109 F: FnOnce(CitadelClientServerConnection<R>) -> Fut + Send,
110 Fut: Future<Output = Result<(), NetworkError>> + Send,
111{
112 fn generate_object_transfer_handle() -> (
113 citadel_io::tokio::sync::mpsc::UnboundedSender<ObjectTransferHandler>,
114 Mutex<Option<FileTransferHandleRx>>,
115 ) {
116 let (tx, rx) = citadel_io::tokio::sync::mpsc::unbounded_channel();
117 let rx = FileTransferHandleRx {
118 inner: rx,
119 conn_type: VirtualTargetType::LocalGroupServer { session_cid: 0 },
120 };
121 (tx, Mutex::new(Some(rx)))
122 }
123
124 pub fn new(settings: ServerConnectionSettings<R>, on_channel_received: F) -> Self {
127 let (udp_mode, session_security_settings) =
128 (settings.udp_mode(), settings.session_security_settings());
129 let server_password = settings.pre_shared_key().cloned();
130 let (tx_incoming_object_transfer_handle, rx_incoming_object_transfer_handle) =
131 Self::generate_object_transfer_handle();
132
133 let connection_type = match settings {
134 ServerConnectionSettings::CredentialedConnect {
135 username, password, ..
136 } => ConnectionType::Connect { username, password },
137
138 ServerConnectionSettings::Transient {
139 server_addr: address,
140 uuid,
141 ..
142 } => ConnectionType::Transient {
143 uuid,
144 server_addr: address,
145 },
146
147 ServerConnectionSettings::CredentialedRegister {
148 alias,
149 username,
150 password,
151 address,
152 ..
153 } => ConnectionType::Register {
154 full_name: alias,
155 server_addr: address,
156 username,
157 password,
158 },
159 };
160
161 Self {
162 handler: Mutex::new(Some(on_channel_received)),
163 udp_mode,
164 auth_info: Mutex::new(Some(connection_type)),
165 session_security_settings,
166 unprocessed_signal_filter_tx: Default::default(),
167 rx_incoming_object_transfer_handle,
168 tx_incoming_object_transfer_handle,
169 server_password,
170 remote: None,
171 _pd: Default::default(),
172 }
173 }
174}
175
176#[async_trait]
177impl<F, Fut, R: Ratchet> NetKernel<R> for SingleClientServerConnectionKernel<F, Fut, R>
178where
179 F: FnOnce(CitadelClientServerConnection<R>) -> Fut + Send,
180 Fut: Future<Output = Result<(), NetworkError>> + Send,
181{
182 fn load_remote(&mut self, server_remote: NodeRemote<R>) -> Result<(), NetworkError> {
183 self.remote = Some(server_remote);
184 Ok(())
185 }
186
187 #[allow(clippy::blocks_in_conditions)]
188 #[cfg_attr(
189 feature = "localhost-testing",
190 tracing::instrument(level = "trace", target = "citadel", skip_all, err(Debug))
191 )]
192 async fn on_start(&self) -> Result<(), NetworkError> {
193 let session_security_settings = self.session_security_settings;
194 let remote = self.remote.clone().unwrap();
195 let (auth_info, handler) = {
196 (
197 self.auth_info.lock().take().unwrap(),
198 self.handler.lock().take().unwrap(),
199 )
200 };
201
202 let auth = match auth_info {
203 ConnectionType::Register {
204 full_name,
205 server_addr,
206 username,
207 password,
208 } => {
209 if !remote
210 .account_manager()
211 .get_persistence_handler()
212 .username_exists(&username)
213 .await?
214 {
215 let _reg_success = remote
216 .register(
217 server_addr,
218 full_name.as_str(),
219 username.as_str(),
220 password.clone(),
221 self.session_security_settings,
222 self.server_password.clone(),
223 )
224 .await?;
225 }
226
227 AuthenticationRequest::credentialed(username, password)
228 }
229
230 ConnectionType::Connect { username, password } => {
231 AuthenticationRequest::credentialed(username, password)
232 }
233
234 ConnectionType::Transient { uuid, server_addr } => {
235 AuthenticationRequest::transient(uuid, server_addr)
236 }
237 };
238
239 let mut connect_success = remote
240 .connect(
241 auth,
242 ConnectMode::Standard { force_login: false },
243 self.udp_mode,
244 None,
245 self.session_security_settings,
246 self.server_password.clone(),
247 )
248 .await?;
249
250 let conn_type = VirtualTargetType::LocalGroupServer {
251 session_cid: connect_success.cid,
252 };
253
254 let mut handle = {
255 let mut lock = self.rx_incoming_object_transfer_handle.lock();
256 lock.take().expect("Should not have been called before")
257 };
258
259 handle.conn_type.set_session_cid(connect_success.cid);
260
261 let (reroute_tx, reroute_rx) = citadel_io::tokio::sync::mpsc::unbounded_channel();
262 *self.unprocessed_signal_filter_tx.lock() = Some(reroute_tx);
263 connect_success.remote = ClientServerRemote::new(
264 conn_type,
265 remote,
266 session_security_settings,
267 Some(reroute_rx),
268 Some(handle),
269 );
270
271 handler(connect_success).await
272 }
273
274 async fn on_node_event_received(&self, message: NodeResult<R>) -> Result<(), NetworkError> {
275 match message {
276 NodeResult::ObjectTransferHandle(handle) => {
277 if let Err(err) = self.tx_incoming_object_transfer_handle.send(handle.handle) {
278 log::warn!(target: "citadel", "failed to send unprocessed NodeResult: {err:?}")
279 }
280 }
281
282 message => {
283 if let Some(val) = self.unprocessed_signal_filter_tx.lock().as_ref() {
284 log::trace!(target: "citadel", "Will forward message {val:?}");
285 if let Err(err) = val.send(message) {
286 log::warn!(target: "citadel", "failed to send unprocessed NodeResult: {err:?}")
287 }
288 }
289 }
290 }
291
292 Ok(())
293 }
294
295 async fn on_stop(&mut self) -> Result<(), NetworkError> {
296 Ok(())
297 }
298}
299
300#[cfg(all(test, feature = "localhost-testing"))]
301mod tests {
302 use crate::prefabs::client::single_connection::SingleClientServerConnectionKernel;
303 use crate::prefabs::client::DefaultServerConnectionSettingsBuilder;
304 use crate::prelude::*;
305 use crate::test_common::{server_info_reactive, wait_for_peers, TestBarrier};
306 use citadel_io::tokio;
307 use rstest::rstest;
308 use std::sync::atomic::{AtomicBool, Ordering};
309 use uuid::Uuid;
310
311 #[cfg_attr(
312 feature = "localhost-testing",
313 tracing::instrument(level = "trace", target = "citadel", skip_all, err(Debug))
314 )]
315 async fn on_server_received_conn<R: Ratchet>(
316 udp_mode: UdpMode,
317 conn: &mut CitadelClientServerConnection<R>,
318 ) -> Result<(), NetworkError> {
319 crate::test_common::udp_mode_assertions(udp_mode, conn.udp_channel_rx.take()).await;
320 Ok(())
321 }
322
323 #[cfg_attr(
324 feature = "localhost-testing",
325 tracing::instrument(level = "trace", target = "citadel", skip_all, err(Debug))
326 )]
327 async fn default_server_harness<R: Ratchet>(
328 udp_mode: UdpMode,
329 mut connection: CitadelClientServerConnection<R>,
330 server_success: &AtomicBool,
331 ) -> Result<(), NetworkError> {
332 wait_for_peers().await;
333 on_server_received_conn(udp_mode, &mut connection).await?;
334 server_success.store(true, Ordering::SeqCst);
335 log::warn!(target: "citadel", "Server awaiting peer ...");
336 wait_for_peers().await;
337 connection.shutdown_kernel().await
338 }
339
340 #[rstest]
341 #[timeout(std::time::Duration::from_secs(90))]
342 #[citadel_io::tokio::test(flavor = "multi_thread")]
343 async fn test_single_connection_registered(
344 #[values(UdpMode::Enabled, UdpMode::Disabled)] udp_mode: UdpMode,
345 #[values(ServerMode::P2P(NativeP2PConfig::self_signed()), ServerMode::OrderedReliableSecure(NativeSecureConfig::self_signed().unwrap())
346 )]
347 underlying_protocol: ServerMode<NativeIO>,
348 ) {
349 citadel_logging::setup_log();
350 TestBarrier::setup(2);
351
352 if cfg!(windows) {
355 match &underlying_protocol {
356 ServerMode::OrderedReliableSecure(..) => {
357 citadel_logging::warn!(target: "citadel", "Skipping TLS test on Windows - self-signed certs may not work");
358 return;
359 }
360 ServerMode::P2P(..) => {
361 citadel_logging::warn!(target: "citadel", "Skipping QUIC test on Windows - socket binding may fail with error 10013");
362 return;
363 }
364 _ => {}
365 }
366 }
367
368 let client_success = &AtomicBool::new(false);
369 let server_success = &AtomicBool::new(false);
370
371 let (server, server_addr) = server_info_reactive::<_, _, StackedRatchet>(
372 move |connection| async move {
373 default_server_harness(udp_mode, connection, server_success).await
374 },
375 |builder| {
376 let _ = builder.with_underlying_protocol(underlying_protocol);
377 },
378 );
379
380 let client_settings = DefaultServerConnectionSettingsBuilder::credentialed_registration(
381 server_addr,
382 "nologik",
383 "Some Alias",
384 "password",
385 )
386 .with_udp_mode(udp_mode)
387 .build()
388 .unwrap();
389
390 let client_kernel =
391 SingleClientServerConnectionKernel::new(client_settings, |mut connection| async move {
392 log::trace!(target: "citadel", "***CLIENT TEST SUCCESS***");
393 let chan = connection.udp_channel_rx.take();
394 wait_for_peers().await;
395 crate::test_common::udp_mode_assertions(udp_mode, chan).await;
396 client_success.store(true, Ordering::Relaxed);
397 wait_for_peers().await;
398 connection.shutdown_kernel().await
399 });
400
401 let client = DefaultNodeBuilder::default()
402 .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
403 .build(client_kernel)
404 .unwrap();
405
406 let joined = futures::future::try_join(server, client);
407
408 let _ = joined.await.expect("Failed to join server and client - possible port binding issue on Windows (error 10013)");
409
410 assert!(client_success.load(Ordering::Relaxed));
411 assert!(server_success.load(Ordering::Relaxed));
412 }
413
414 #[rstest]
415 #[case(UdpMode::Enabled, None, HeaderObfuscatorSettings::Disabled)]
416 #[case(UdpMode::Enabled, None, HeaderObfuscatorSettings::Enabled)]
417 #[case(
418 UdpMode::Enabled,
419 None,
420 HeaderObfuscatorSettings::EnabledWithKey(12345)
421 )]
422 #[case(
423 UdpMode::Enabled,
424 Some("test-password"),
425 HeaderObfuscatorSettings::Disabled
426 )]
427 #[timeout(std::time::Duration::from_secs(90))]
428 #[citadel_io::tokio::test(flavor = "multi_thread")]
429 async fn test_single_connection_transient(
430 #[case] udp_mode: UdpMode,
431 #[case] server_password: Option<&'static str>,
432 #[case] header_obfuscator_settings: HeaderObfuscatorSettings,
433 ) {
434 citadel_logging::setup_log();
435 TestBarrier::setup(2);
436
437 let client_success = &AtomicBool::new(false);
438 let server_success = &AtomicBool::new(false);
439 let (server, server_addr) = server_info_reactive::<_, _, StackedRatchet>(
440 |connection| async move {
441 default_server_harness(udp_mode, connection, server_success).await
442 },
443 |opts| {
444 if let Some(password) = server_password {
445 let _ = opts
446 .with_server_password(password)
447 .with_server_declared_header_obfuscation(header_obfuscator_settings);
448 }
449 },
450 );
451
452 let uuid = Uuid::new_v4();
453
454 let mut server_connection_settings =
455 DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
456 .with_udp_mode(udp_mode)
457 .with_session_security_settings(SessionSecuritySettings {
458 security_level: Default::default(),
459 secrecy_mode: Default::default(),
460 crypto_params: Default::default(),
461 header_obfuscator_settings,
462 });
463
464 if let Some(server_password) = server_password {
465 server_connection_settings =
466 server_connection_settings.with_session_password(server_password);
467 }
468
469 let server_connection_settings = server_connection_settings.build().unwrap();
470
471 let client_kernel = SingleClientServerConnectionKernel::new(
472 server_connection_settings,
473 |mut connection| async move {
474 log::trace!(target: "citadel", "***CLIENT TEST SUCCESS***");
475 let chan = connection.udp_channel_rx.take();
476 wait_for_peers().await;
477 crate::test_common::udp_mode_assertions(udp_mode, chan).await;
478 let sessions = connection.remote.sessions().await?;
480 assert!(!sessions.sessions.is_empty());
481 connection.disconnect().await?;
482 client_success.store(true, Ordering::Relaxed);
483 wait_for_peers().await;
484 connection.shutdown_kernel().await
485 },
486 );
487
488 let client = DefaultNodeBuilder::default()
489 .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
490 .build(client_kernel)
491 .unwrap();
492
493 let joined = futures::future::try_join(server, client);
494
495 let _ = joined.await.expect("Failed to join server and client - possible port binding issue on Windows (error 10013)");
496
497 assert!(client_success.load(Ordering::Relaxed));
498 assert!(server_success.load(Ordering::Relaxed));
499 }
500
501 #[rstest]
502 #[case(UdpMode::Enabled, Some("test-password"))]
503 #[timeout(std::time::Duration::from_secs(90))]
504 #[tokio::test(flavor = "multi_thread")]
505 async fn test_single_connection_transient_wrong_password(
506 #[case] udp_mode: UdpMode,
507 #[case] server_password: Option<&'static str>,
508 ) {
509 citadel_logging::setup_log();
510 TestBarrier::setup(2);
511
512 let (server, server_addr) = server_info_reactive::<_, _, StackedRatchet>(
513 |_conn| async move { panic!("Server should not have connected") },
514 |opts| {
515 if let Some(password) = server_password {
516 let _ = opts.with_server_password(password);
517 }
518 },
519 );
520
521 let uuid = Uuid::new_v4();
522
523 let server_connection_settings =
524 DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
525 .with_udp_mode(udp_mode)
526 .with_session_password("wrong-password")
527 .build()
528 .unwrap();
529
530 let client_kernel = SingleClientServerConnectionKernel::new(
531 server_connection_settings,
532 |_connection| async move { panic!("Client should not have connected") },
533 );
534
535 let client = DefaultNodeBuilder::default()
536 .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
537 .build(client_kernel)
538 .unwrap();
539
540 tokio::select! {
541 _res0 = server => {
542 panic!("Server should never finish")
543 },
544
545 result = client => {
546 if let Err(error) = result {
547 assert!(error.into_string().contains("Encryption failure"));
550 } else {
551 panic!("Client should not have connected")
552 }
553 }
554 }
555 }
556
557 #[rstest]
558 #[case(UdpMode::Disabled)]
559 #[timeout(std::time::Duration::from_secs(90))]
560 #[citadel_io::tokio::test(flavor = "multi_thread")]
561 async fn test_single_connection_transient_deregister(#[case] udp_mode: UdpMode) {
562 citadel_logging::setup_log();
563 TestBarrier::setup(2);
564
565 let client_success = &AtomicBool::new(false);
566 let server_success = &AtomicBool::new(false);
567
568 let (server, server_addr) = server_info_reactive::<_, _, StackedRatchet>(
569 |connection| async move {
570 default_server_harness(udp_mode, connection, server_success).await
571 },
572 |_| (),
573 );
574
575 let uuid = Uuid::new_v4();
576
577 let server_connection_settings =
578 DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
579 .with_udp_mode(udp_mode)
580 .build()
581 .unwrap();
582
583 let client_kernel = SingleClientServerConnectionKernel::new(
584 server_connection_settings,
585 |mut connection| async move {
586 log::trace!(target: "citadel", "***CLIENT TEST SUCCESS***");
587 let chan = connection.udp_channel_rx.take();
588 wait_for_peers().await;
589 crate::test_common::udp_mode_assertions(udp_mode, chan).await;
590 connection.deregister().await?;
591 client_success.store(true, Ordering::Relaxed);
592 wait_for_peers().await;
593 connection.shutdown_kernel().await
594 },
595 );
596
597 let client = DefaultNodeBuilder::default()
598 .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
599 .build(client_kernel)
600 .unwrap();
601
602 let joined = futures::future::try_join(server, client);
603
604 let _ = joined.await.expect("Failed to join server and client - possible port binding issue on Windows (error 10013)");
605
606 assert!(client_success.load(Ordering::Relaxed));
607 assert!(server_success.load(Ordering::Relaxed));
608 }
609
610 #[rstest]
611 #[timeout(std::time::Duration::from_secs(90))]
612 #[citadel_io::tokio::test(flavor = "multi_thread")]
613 async fn test_backend_store_c2s() {
614 citadel_logging::setup_log();
615 TestBarrier::setup(2);
616
617 let udp_mode = UdpMode::Disabled;
618
619 let client_success = &AtomicBool::new(false);
620 let server_success = &AtomicBool::new(false);
621 let (server, server_addr) = server_info_reactive::<_, _, StackedRatchet>(
622 |connection| async move {
623 default_server_harness(udp_mode, connection, server_success).await
624 },
625 |_| (),
626 );
627
628 let uuid = Uuid::new_v4();
629
630 let server_connection_settings =
631 DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
632 .with_udp_mode(udp_mode)
633 .build()
634 .unwrap();
635
636 let client_kernel = SingleClientServerConnectionKernel::new(
637 server_connection_settings,
638 |connection| async move {
639 log::trace!(target: "citadel", "***CLIENT TEST SUCCESS***");
640 wait_for_peers().await;
641
642 const KEY: &str = "HELLO_WORLD";
643 const KEY2: &str = "HELLO_WORLD2";
644 let value: Vec<u8> = Vec::from("Hello, world!");
645 let value2: Vec<u8> = Vec::from("Hello, world!2");
646
647 assert_eq!(connection.set(KEY, value.clone()).await?.as_deref(), None);
648 assert_eq!(
649 connection.get(KEY).await?.as_deref(),
650 Some(value.as_slice())
651 );
652
653 assert_eq!(connection.set(KEY2, value2.clone()).await?.as_deref(), None);
654 assert_eq!(
655 connection.get(KEY2).await?.as_deref(),
656 Some(value2.as_slice())
657 );
658
659 let map = connection.get_all().await?;
660 assert_eq!(map.get(KEY), Some(&value));
661 assert_eq!(map.get(KEY2), Some(&value2));
662
663 assert_eq!(
664 connection.remove(KEY2).await?.as_deref(),
665 Some(value2.as_slice())
666 );
667
668 assert_eq!(connection.remove(KEY2).await?.as_deref(), None);
669
670 let map = connection.remove_all().await?;
671 assert_eq!(map.get(KEY), Some(&value));
672 assert_eq!(map.get(KEY2), None);
673
674 assert_eq!(connection.get_all().await?.len(), 0);
675 assert_eq!(connection.remove_all().await?.len(), 0);
676
677 client_success.store(true, Ordering::Relaxed);
678 wait_for_peers().await;
679 connection.shutdown_kernel().await
680 },
681 );
682
683 let client = DefaultNodeBuilder::default()
684 .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
685 .build(client_kernel)
686 .unwrap();
687
688 let joined = futures::future::try_join(server, client);
689
690 let _ = joined.await.expect("Failed to join server and client - possible port binding issue on Windows (error 10013)");
691
692 assert!(client_success.load(Ordering::Relaxed));
693 assert!(server_success.load(Ordering::Relaxed));
694 }
695
696 #[rstest]
697 #[timeout(std::time::Duration::from_secs(90))]
698 #[citadel_io::tokio::test(flavor = "multi_thread")]
699 async fn test_rekey_c2s() {
700 citadel_logging::setup_log();
701 TestBarrier::setup(2);
702
703 let udp_mode = UdpMode::Disabled;
704
705 let client_success = &AtomicBool::new(false);
706 let server_success = &AtomicBool::new(false);
707 let (server, server_addr) = server_info_reactive::<_, _, StackedRatchet>(
708 |connection| async move {
709 default_server_harness(udp_mode, connection, server_success).await
710 },
711 |_| (),
712 );
713
714 let uuid = Uuid::new_v4();
715
716 let server_connection_settings =
717 DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
718 .with_udp_mode(udp_mode)
719 .build()
720 .unwrap();
721
722 let client_kernel = SingleClientServerConnectionKernel::new(
723 server_connection_settings,
724 |mut connection| async move {
725 log::trace!(target: "citadel", "***CLIENT LOGIN SUCCESS***");
726 wait_for_peers().await;
727 let chan = connection.udp_channel_rx.take();
728 crate::test_common::udp_mode_assertions(udp_mode, chan).await;
729
730 for x in 1..10 {
731 assert_eq!(connection.remote.rekey().await?, Some(x));
732 }
733
734 client_success.store(true, Ordering::Relaxed);
735 wait_for_peers().await;
736
737 connection.shutdown_kernel().await
738 },
739 );
740
741 let client = DefaultNodeBuilder::default()
742 .with_stun_servers(crate::test_common::LOCALHOST_STUN_SERVERS)
743 .build(client_kernel)
744 .unwrap();
745 let joined = futures::future::try_join(server, client);
746
747 let _ = joined.await.expect("Failed to join server and client - possible port binding issue on Windows (error 10013)");
748
749 assert!(client_success.load(Ordering::Relaxed));
750 assert!(server_success.load(Ordering::Relaxed));
751 }
752}