citadel_sdk/

fs.rs

//! Remote Encrypted Virtual Filesystem (RE-VFS)
//!
//! This module provides high-level operations for interacting with the Remote Encrypted
//! Virtual Filesystem in the Citadel Protocol. RE-VFS enables secure storage and
//! retrieval of files with end-to-end encryption.
//!
//! # Features
//! - Secure file storage and retrieval
//! - Configurable security levels
//! - Automatic encryption handling
//! - Virtual path management
//! - File deletion support
//! - Take operations for atomic reads
//!
//! # Example
//! ```rust
//! use citadel_sdk::prelude::*;
//! use citadel_sdk::fs;
//!
//! async fn store_file<R: Ratchet>(remote: &impl TargetLockedRemote<R>) -> Result<(), NetworkError> {
//!     // Write a file to RE-VFS
//!     fs::write(remote, "/local/file.txt", "/virtual/file.txt").await?;
//!
//!     // Read the file back
//!     let local_path = fs::read(remote, "/virtual/file.txt").await?;
//!
//!     // Delete the file
//!     fs::delete(remote, "/virtual/file.txt").await?;
//!
//!     Ok(())
//! }
//! ```
//!
//! # Important Notes
//! - All operations are end-to-end encrypted
//! - Security levels are configurable per operation
//! - Take operations atomically read and delete
//! - Virtual paths are independent of local paths
//!
//! # Related Components
//! - [`TargetLockedRemote`]: Remote connection interface
//! - [`SecurityLevel`]: Encryption strength configuration
//! - [`ObjectSource`]: File source abstraction
//! - [`NetworkError`]: Error handling
//!
//! [`TargetLockedRemote`]: crate::prelude::TargetLockedRemote
//! [`SecurityLevel`]: crate::prelude::SecurityLevel
//! [`ObjectSource`]: crate::prelude::ObjectSource
//! [`NetworkError`]: crate::prelude::NetworkError
//!

use crate::prelude::{ObjectSource, ProtocolRemoteTargetExt, TargetLockedRemote};

use citadel_proto::prelude::NetworkError;
use citadel_proto::prelude::*;
use citadel_types::crypto::SecurityLevel;
use std::path::PathBuf;

/// Writes a file or BytesSource to the Remote Encrypted Virtual Filesystem
pub async fn write<T: ObjectSource, P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    source: T,
    virtual_path: P,
) -> Result<(), NetworkError> {
    write_with_security_level(remote, source, Default::default(), virtual_path).await
}

/// Writes a file or BytesSource to the Remote Encrypted Virtual Filesystem with a custom security level.
pub async fn write_with_security_level<T: ObjectSource, P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    source: T,
    security_level: SecurityLevel,
    virtual_path: P,
) -> Result<(), NetworkError> {
    remote
        .remote_encrypted_virtual_filesystem_push(source, virtual_path, security_level)
        .await
}

/// Reads a file from the Remote Encrypted Virtual Filesystem
pub async fn read<P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    virtual_path: P,
) -> Result<PathBuf, NetworkError> {
    read_with_security_level(remote, Default::default(), virtual_path).await
}

/// Reads a file from the Remote Encrypted Virtual Filesystem with a custom transport security level
pub async fn read_with_security_level<P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    transfer_security_level: SecurityLevel,
    virtual_path: P,
) -> Result<PathBuf, NetworkError> {
    remote
        .remote_encrypted_virtual_filesystem_pull(virtual_path, transfer_security_level, false)
        .await
}

/// Takes a file from the Remote Encrypted Virtual Filesystem
pub async fn take<P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    virtual_path: P,
) -> Result<PathBuf, NetworkError> {
    remote
        .remote_encrypted_virtual_filesystem_pull(virtual_path, Default::default(), true)
        .await
}

/// Takes a file from the Remote Encrypted Virtual Filesystem with a custom security level.
pub async fn take_with_security_level<P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    transfer_security_level: SecurityLevel,
    virtual_path: P,
) -> Result<PathBuf, NetworkError> {
    remote
        .remote_encrypted_virtual_filesystem_pull(virtual_path, transfer_security_level, true)
        .await
}

/// Deletes a file from the Remote Encrypted Virtual Filesystem
pub async fn delete<P: Into<PathBuf> + Send, R: Ratchet>(
    remote: &impl TargetLockedRemote<R>,
    virtual_path: P,
) -> Result<(), NetworkError> {
    remote
        .remote_encrypted_virtual_filesystem_delete(virtual_path)
        .await
}

#[cfg(test)]
mod tests {
    use crate::prefabs::client::single_connection::SingleClientServerConnectionKernel;
    use crate::prefabs::server::accept_file_transfer_kernel::AcceptFileTransferKernel;

    use crate::prefabs::client::peer_connection::FileTransferHandleRx;
    #[cfg(feature = "localhost-testing")]
    use crate::prefabs::client::peer_connection::PeerConnectionKernel;
    use crate::prefabs::client::DefaultServerConnectionSettingsBuilder;
    use crate::prelude::*;
    #[cfg(feature = "localhost-testing")]
    use crate::test_common::wait_for_peers;
    use citadel_io::tokio;
    use futures::StreamExt;
    use rstest::rstest;
    use std::net::SocketAddr;
    use std::path::PathBuf;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::time::Duration;
    use uuid::Uuid;

    pub fn server_info<'a, R: Ratchet>() -> (NodeFuture<'a, AcceptFileTransferKernel<R>>, SocketAddr)
    {
        crate::test_common::server_test_node(AcceptFileTransferKernel::<R>::default(), |_| {})
    }

    #[rstest]
    #[case(
        EncryptionAlgorithm::AES_GCM_256,
        KemAlgorithm::Kyber,
        SigAlgorithm::None
    )]
    #[case(
        EncryptionAlgorithm::KyberHybrid,
        KemAlgorithm::Kyber,
        SigAlgorithm::Dilithium65
    )]
    #[timeout(Duration::from_secs(90))]
    #[citadel_io::tokio::test]
    async fn test_c2s_file_transfer_revfs(
        #[case] enx: EncryptionAlgorithm,
        #[case] kem: KemAlgorithm,
        #[case] sig: SigAlgorithm,
        #[values(SecurityLevel::Standard, SecurityLevel::Reinforced)] security_level: SecurityLevel,
    ) {
        citadel_logging::setup_log();
        let client_success = &AtomicBool::new(false);
        let (server, server_addr) = server_info::<StackedRatchet>();
        let uuid = Uuid::new_v4();

        let source_dir = PathBuf::from("../resources/TheBridge.pdf");

        let session_security_settings = SessionSecuritySettingsBuilder::default()
            .with_crypto_params(enx + kem + sig)
            .with_security_level(security_level)
            .build()
            .unwrap();

        let server_connection_settings =
            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
                .disable_udp()
                .with_session_security_settings(session_security_settings)
                .build()
                .unwrap();

        let client_kernel = SingleClientServerConnectionKernel::new(
            server_connection_settings,
            |connection| async move {
                log::trace!(target: "citadel", "***CLIENT LOGIN SUCCESS :: File transfer next ***");

                // Test list_sessions() - verify we can list active sessions with correct data
                let sessions = connection.remote.list_sessions().await?;
                log::info!(target: "citadel", "***CLIENT list_sessions() returned {} sessions***", sessions.sessions.len());
                assert!(
                    !sessions.sessions.is_empty(),
                    "Should have at least one active session"
                );
                // Find our session by CID
                let our_session = sessions.sessions.iter().find(|s| s.cid == connection.cid);
                assert!(our_session.is_some(), "Should find our session in the list");
                let our_session = our_session.unwrap();
                assert!(
                    !our_session.connections.is_empty(),
                    "Session should have at least one connection"
                );
                // For C2S, peer_cid should be None
                let c2s_conn = our_session
                    .connections
                    .iter()
                    .find(|c| c.peer_cid.is_none());
                assert!(
                    c2s_conn.is_some(),
                    "Should have a C2S connection (peer_cid = None)"
                );
                let c2s_conn = c2s_conn.unwrap();
                assert!(c2s_conn.connected, "C2S connection should be active");
                log::info!(target: "citadel", "***CLIENT list_sessions() verification PASSED***");

                let virtual_path = PathBuf::from("/home/john.doe/TheBridge.pdf");
                // write to file to the RE-VFS
                crate::fs::write_with_security_level(
                    &connection.remote,
                    source_dir.clone(),
                    security_level,
                    &virtual_path,
                )
                .await?;
                log::info!(target: "citadel", "***CLIENT FILE TRANSFER SUCCESS***");
                // now, pull it
                let save_dir = crate::fs::read(&connection.remote, virtual_path).await?;
                // now, compare bytes
                log::info!(target: "citadel", "***CLIENT REVFS PULL SUCCESS");
                let original_bytes = citadel_io::tokio::fs::read(&source_dir).await.unwrap();
                let revfs_pulled_bytes = citadel_io::tokio::fs::read(&save_dir).await.unwrap();
                assert_eq!(original_bytes, revfs_pulled_bytes);
                log::info!(target: "citadel", "***CLIENT REVFS PULL COMPARE SUCCESS");
                client_success.store(true, Ordering::Relaxed);
                connection.shutdown_kernel().await
            },
        );

        let client = DefaultNodeBuilder::default().build(client_kernel).unwrap();

        let result = citadel_io::tokio::select! {
            res0 = client => res0.map(|_| ()),
            res1 = server => res1.map(|_| ())
        };

        result.unwrap();

        assert!(client_success.load(Ordering::Relaxed));
    }

    #[rstest]
    #[case(
        EncryptionAlgorithm::AES_GCM_256,
        KemAlgorithm::Kyber,
        SigAlgorithm::None
    )]
    #[timeout(std::time::Duration::from_secs(90))]
    #[citadel_io::tokio::test]
    async fn test_c2s_file_transfer_revfs_take(
        #[case] enx: EncryptionAlgorithm,
        #[case] kem: KemAlgorithm,
        #[case] sig: SigAlgorithm,
        #[values(SecurityLevel::Standard)] security_level: SecurityLevel,
    ) {
        citadel_logging::setup_log();
        let client_success = &AtomicBool::new(false);
        let (server, server_addr) = server_info::<StackedRatchet>();
        let uuid = Uuid::new_v4();

        let source_dir = PathBuf::from("../resources/TheBridge.pdf");

        let session_security_settings = SessionSecuritySettingsBuilder::default()
            .with_crypto_params(enx + kem + sig)
            .with_security_level(security_level)
            .build()
            .unwrap();

        let server_connection_settings =
            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
                .disable_udp()
                .with_session_security_settings(session_security_settings)
                .build()
                .unwrap();

        let client_kernel = SingleClientServerConnectionKernel::new(
            server_connection_settings,
            |connection| async move {
                log::trace!(target: "citadel", "***CLIENT LOGIN SUCCESS :: File transfer next ***");
                let virtual_path = PathBuf::from("/home/john.doe/TheBridge.pdf");
                // write to file to the RE-VFS
                crate::fs::write_with_security_level(
                    &connection.remote,
                    source_dir.clone(),
                    security_level,
                    &virtual_path,
                )
                .await?;
                log::trace!(target: "citadel", "***CLIENT FILE TRANSFER SUCCESS***");
                // now, pull it
                let save_dir = crate::fs::take(&connection.remote, &virtual_path).await?;
                // now, compare bytes
                log::trace!(target: "citadel", "***CLIENT REVFS PULL SUCCESS");
                let original_bytes = citadel_io::tokio::fs::read(&source_dir).await.unwrap();
                let revfs_pulled_bytes = citadel_io::tokio::fs::read(&save_dir).await.unwrap();
                assert_eq!(original_bytes, revfs_pulled_bytes);
                log::trace!(target: "citadel", "***CLIENT REVFS PULL COMPARE SUCCESS");
                // prove we can no longer read from this virtual file
                assert!(crate::fs::read(&connection.remote, &virtual_path)
                    .await
                    .is_err());
                client_success.store(true, Ordering::Relaxed);
                connection.shutdown_kernel().await
            },
        );

        let client = DefaultNodeBuilder::default().build(client_kernel).unwrap();

        let result = citadel_io::tokio::select! {
            res0 = client => res0.map(|_| ()),
            res1 = server => res1.map(|_| ())
        };

        result.unwrap();

        assert!(client_success.load(Ordering::Relaxed));
    }

    #[rstest]
    #[case(
        EncryptionAlgorithm::AES_GCM_256,
        KemAlgorithm::Kyber,
        SigAlgorithm::None
    )]
    #[timeout(std::time::Duration::from_secs(90))]
    #[citadel_io::tokio::test]
    async fn test_c2s_file_transfer_revfs_delete(
        #[case] enx: EncryptionAlgorithm,
        #[case] kem: KemAlgorithm,
        #[case] sig: SigAlgorithm,
        #[values(SecurityLevel::Standard)] security_level: SecurityLevel,
    ) {
        citadel_logging::setup_log();
        let client_success = &AtomicBool::new(false);
        let (server, server_addr) = server_info::<StackedRatchet>();
        let uuid = Uuid::new_v4();

        let source_dir = PathBuf::from("../resources/TheBridge.pdf");

        let session_security_settings = SessionSecuritySettingsBuilder::default()
            .with_crypto_params(enx + kem + sig)
            .with_security_level(security_level)
            .build()
            .unwrap();

        let server_connection_settings =
            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid)
                .disable_udp()
                .with_session_security_settings(session_security_settings)
                .build()
                .unwrap();

        let client_kernel = SingleClientServerConnectionKernel::new(
            server_connection_settings,
            |connection| async move {
                log::trace!(target: "citadel", "***CLIENT LOGIN SUCCESS :: File transfer next ***");
                let virtual_path = PathBuf::from("/home/john.doe/TheBridge.pdf");
                // write to file to the RE-VFS
                crate::fs::write_with_security_level(
                    &connection.remote,
                    source_dir.clone(),
                    security_level,
                    &virtual_path,
                )
                .await?;
                log::trace!(target: "citadel", "***CLIENT FILE TRANSFER SUCCESS***");
                // now, pull it
                let save_dir = crate::fs::read(&connection.remote, &virtual_path).await?;
                // now, compare bytes
                log::trace!(target: "citadel", "***CLIENT REVFS PULL SUCCESS");
                let original_bytes = citadel_io::tokio::fs::read(&source_dir).await.unwrap();
                let revfs_pulled_bytes = citadel_io::tokio::fs::read(&save_dir).await.unwrap();
                assert_eq!(original_bytes, revfs_pulled_bytes);
                log::trace!(target: "citadel", "***CLIENT REVFS PULL COMPARE SUCCESS");
                crate::fs::delete(&connection.remote, &virtual_path).await?;
                // prove we can no longer read from this virtual file since it was just deleted
                assert!(crate::fs::read(&connection.remote, &virtual_path)
                    .await
                    .is_err());
                client_success.store(true, Ordering::Relaxed);
                connection.shutdown_kernel().await
            },
        );

        let client = DefaultNodeBuilder::default().build(client_kernel).unwrap();

        let result = citadel_io::tokio::select! {
            res0 = client => res0.map(|_| ()),
            res1 = server => res1.map(|_| ())
        };

        result.unwrap();

        assert!(client_success.load(Ordering::Relaxed));
    }

    #[rstest]
    #[case(SecrecyMode::BestEffort)]
    #[timeout(Duration::from_secs(60))]
    #[cfg(feature = "localhost-testing")]
    #[citadel_io::tokio::test(flavor = "multi_thread")]
    async fn test_p2p_file_transfer_revfs(
        #[case] secrecy_mode: SecrecyMode,
        #[values(KemAlgorithm::Kyber)] kem: KemAlgorithm,
        #[values(EncryptionAlgorithm::AES_GCM_256)] enx: EncryptionAlgorithm,
    ) {
        citadel_logging::setup_log();
        crate::test_common::TestBarrier::setup(2);
        let client0_success = &AtomicBool::new(false);
        let client1_success = &AtomicBool::new(false);

        let (server, server_addr) = crate::test_common::server_info::<StackedRatchet>();

        let uuid0 = Uuid::new_v4();
        let uuid1 = Uuid::new_v4();
        let session_security = SessionSecuritySettingsBuilder::default()
            .with_secrecy_mode(secrecy_mode)
            .with_crypto_params(kem + enx)
            .build()
            .unwrap();

        let security_level = SecurityLevel::Standard;

        let source_dir = &PathBuf::from("../resources/TheBridge.pdf");

        let server_connection_settings =
            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid0)
                .disable_udp()
                .with_session_security_settings(session_security)
                .build()
                .unwrap();

        let peer_conn_0 = PeerConnectionSetupAggregator::default()
            .with_peer_custom(uuid1)
            .ensure_registered()
            .with_session_security_settings(session_security)
            .enable_udp()
            .add();

        // TODO: SinglePeerConnectionKernel
        let client_kernel0 = PeerConnectionKernel::new(
            server_connection_settings,
            peer_conn_0,
            move |mut connection, remote_outer| async move {
                wait_for_peers().await;
                let mut connection = connection.recv().await.unwrap()?;
                let cid = connection.channel.get_session_cid();
                wait_for_peers().await;
                // The other peer will send the file first
                log::info!(target: "citadel", "***CLIENT A {cid} LOGIN SUCCESS :: File transfer next ***");
                let remote = connection.remote.clone();
                let handle_orig = connection.incoming_object_transfer_handles.take().unwrap();
                let _file_transfer_task = accept_all(handle_orig);

                let virtual_path = PathBuf::from("/home/john.doe/TheBridge.pdf");
                // write the file to the RE-VFS
                crate::fs::write_with_security_level(
                    &remote,
                    source_dir.clone(),
                    security_level,
                    &virtual_path,
                )
                .await?;
                log::info!(target: "citadel", "***CLIENT A {cid} FILE TRANSFER SUCCESS***");
                tokio::time::sleep(Duration::from_secs(1)).await;
                wait_for_peers().await;
                // now, pull it
                let save_dir = crate::fs::read(&remote, virtual_path).await?;
                // now, compare bytes
                log::info!(target: "citadel", "***CLIENT A {cid} REVFS PULL SUCCESS");
                let original_bytes = tokio::fs::read(&source_dir).await.unwrap();
                let revfs_pulled_bytes = tokio::fs::read(&save_dir).await.unwrap();
                assert_eq!(original_bytes, revfs_pulled_bytes);
                log::info!(target: "citadel", "***CLIENT A {cid} REVFS PULL COMPARE SUCCESS");
                wait_for_peers().await;
                client0_success.store(true, Ordering::Relaxed);
                remote_outer.shutdown_kernel().await
            },
        );

        let server_connection_settings =
            DefaultServerConnectionSettingsBuilder::transient_with_id(server_addr, uuid1)
                .disable_udp()
                .with_session_security_settings(session_security)
                .build()
                .unwrap();

        let peer_conn_1 = PeerConnectionSetupAggregator::default()
            .with_peer_custom(uuid0)
            .ensure_registered()
            .with_session_security_settings(session_security)
            .enable_udp()
            .add();

        let client_kernel1 = PeerConnectionKernel::new(
            server_connection_settings,
            peer_conn_1,
            move |mut connection, remote_outer| async move {
                wait_for_peers().await;
                let mut connection = connection.recv().await.unwrap()?;
                let cid = connection.channel.get_session_cid();
                wait_for_peers().await;
                let remote = connection.remote.clone();
                let handle_orig = connection.incoming_object_transfer_handles.take().unwrap();
                let _file_transfer_task = accept_all(handle_orig);
                log::info!(target: "citadel", "***CLIENT B {cid} LOGIN SUCCESS :: File transfer next ***");
                let virtual_path = PathBuf::from("/home/john.doe/TheBridge.pdf");
                // write the file to the RE-VFS
                crate::fs::write_with_security_level(
                    &remote,
                    source_dir.clone(),
                    security_level,
                    &virtual_path,
                )
                .await?;
                log::info!(target: "citadel", "***CLIENT B {cid} FILE TRANSFER SUCCESS***");
                // Wait some time for the file to synchronize
                tokio::time::sleep(Duration::from_secs(1)).await;
                wait_for_peers().await;
                // now, pull it
                let save_dir = crate::fs::read(&remote, virtual_path).await?;
                // now, compare bytes
                log::info!(target: "citadel", "***CLIENT B {cid} REVFS PULL SUCCESS");
                let original_bytes = citadel_io::tokio::fs::read(&source_dir).await.unwrap();
                let revfs_pulled_bytes = citadel_io::tokio::fs::read(&save_dir).await.unwrap();
                assert_eq!(original_bytes, revfs_pulled_bytes);
                log::info!(target: "citadel", "***CLIENT B {cid} REVFS PULL COMPARE SUCCESS");
                wait_for_peers().await;
                client1_success.store(true, Ordering::Relaxed);
                remote_outer.shutdown_kernel().await
            },
        );

        let client0 = DefaultNodeBuilder::default().build(client_kernel0).unwrap();
        let client1 = DefaultNodeBuilder::default().build(client_kernel1).unwrap();
        let clients = futures::future::try_join(client0, client1);

        let task = async move {
            citadel_io::tokio::select! {
                server_res = server => Err(NetworkError::msg(format!("Server ended prematurely: {:?}", server_res.map(|_| ())))),
                client_res = clients => client_res.map(|_| ())
            }
        };

        let _ = citadel_io::tokio::time::timeout(Duration::from_secs(120), task)
            .await
            .unwrap();

        assert!(client0_success.load(Ordering::Relaxed));
        assert!(client1_success.load(Ordering::Relaxed));
    }

    #[allow(dead_code)]
    fn accept_all(mut rx: FileTransferHandleRx) -> citadel_io::tokio::task::JoinHandle<()> {
        citadel_io::tokio::task::spawn(async move {
            while let Some(mut handle) = rx.recv().await {
                if let Err(err) = handle.accept() {
                    log::error!(target: "citadel", "Failed to accept file transfer: {err:?}");
                    continue;
                }

                // Wait for this transfer to complete before accepting the next one
                exhaust_file_transfer_async(handle).await;
            }
        })
    }

    #[allow(dead_code)]
    async fn exhaust_file_transfer_async(mut handle: ObjectTransferHandler) {
        while let Some(evt) = handle.next().await {
            log::info!(target: "citadel", "File Transfer Event: {evt:?}");
            if let ObjectTransferStatus::Fail(err) = &evt {
                log::error!(target: "citadel", "File Transfer Failed: {err:?}");
                break;
            } else if let ObjectTransferStatus::TransferComplete = &evt {
                break;
            } else if let ObjectTransferStatus::ReceptionComplete = &evt {
                break;
            }
        }
    }
}