Blocktree/crates/btfuse/src/main.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
mod fuse_daemon;
use btconfig::{CredStoreConfig, FigmentExt, NodeCredConsumer};
use figment::{providers::Serialized, Figment};
use fuse_daemon::FuseDaemon;
mod fuse_fs;

use btfproto::{client::FsClient, local_fs::LocalFs, server::FsProvider};
use btlib::{
    bterr,
    crypto::{Creds, CredsPriv},
    Result,
};
use btmsg::{BlockAddr, Transmitter};
use serde::{Deserialize, Serialize};
use std::{
    fs::{self},
    io,
    num::NonZeroUsize,
    path::{Path, PathBuf},
    sync::Arc,
};
use tokio::sync::oneshot;

#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum FsKind {
    /// Indicates that a local filesystem provider should be used.
    Local { path: PathBuf },
    /// Indicates that a remote filesystem provider should be used.
    Remote { addr: BlockAddr },
}

impl Default for FsKind {
    fn default() -> Self {
        Self::Local {
            path: btconfig::default_block_dir(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BtfuseConfig {
    /// Configures the credential store used.
    #[serde(rename = "credstore")]
    pub cred_store: CredStoreConfig,
    /// Configures the filesystem provider used.
    #[serde(rename = "fskind")]
    pub fs_kind: FsKind,
    /// Sets the directory where the filesystem will be mounted.
    #[serde(rename = "mntdir")]
    pub mnt_dir: PathBuf,
    /// Specifies the options to use for the filesystem mount.
    #[serde(rename = "mntoptions")]
    pub mnt_options: String,
    /// Prescribes the number of threads to use for handling FUSE messages from the kernel. If
    /// not specified, then the number of threads will be equal to the number of logical cores on
    /// the system.
    pub threads: Option<NonZeroUsize>,
}

impl BtfuseConfig {
    pub fn new() -> Result<Self> {
        Figment::new()
            .merge(Serialized::defaults(BtfuseConfig::default()))
            .btconfig()?
            .extract()
            .map_err(|err| err.into())
    }
}

impl Default for BtfuseConfig {
    fn default() -> Self {
        Self {
            cred_store: CredStoreConfig::default(),
            fs_kind: FsKind::default(),
            mnt_dir: "./btfuse_mnt".into(),
            mnt_options: "default_permissions".into(),
            threads: None,
        }
    }
}

trait PathExt {
    fn try_create_dir(&self) -> io::Result<()>;
}

impl<T: AsRef<Path>> PathExt for T {
    fn try_create_dir(&self) -> io::Result<()> {
        match fs::create_dir(self) {
            Ok(_) => Ok(()),
            Err(err) => match err.kind() {
                io::ErrorKind::AlreadyExists => Ok(()),
                _ => Err(err),
            },
        }
    }
}

async fn local_provider(btdir: PathBuf, node_creds: Arc<dyn Creds>) -> Result<impl FsProvider> {
    btdir.try_create_dir()?;
    let empty = fs::read_dir(&btdir)?.next().is_none();
    if empty {
        LocalFs::new_empty(btdir, 0, node_creds, btfproto::local_fs::ModeAuthorizer {}).await
    } else {
        LocalFs::new_existing(btdir, node_creds, btfproto::local_fs::ModeAuthorizer {})
    }
}

async fn remote_provider<C: 'static + Creds + Send + Sync>(
    remote_addr: BlockAddr,
    node_creds: C,
) -> Result<impl FsProvider> {
    let tx = Transmitter::new(Arc::new(remote_addr), Arc::new(node_creds)).await?;
    let client = FsClient::new(tx);
    Ok(client)
}

async fn run_daemon(config: BtfuseConfig, mounted_signal: Option<oneshot::Sender<()>>) {
    let node_creds = config
        .cred_store
        .consume(NodeCredConsumer)
        .unwrap()
        .unwrap();
    let fallback_path = {
        let writecap = node_creds
            .writecap()
            .ok_or(btlib::BlockError::MissingWritecap)
            .unwrap();
        Arc::new(writecap.bind_path())
    };
    let mut daemon = match config.fs_kind {
        FsKind::Local { path: btdir } => {
            log::info!("starting daemon with local provider using {:?}", btdir);
            let provider = local_provider(btdir.clone(), node_creds)
                .await
                .map_err(|err| {
                    bterr!(
                        "failed to create local provider using path '{}': {err}",
                        btdir.display()
                    )
                })
                .unwrap();
            FuseDaemon::new(
                config.mnt_dir,
                &config.mnt_options,
                config.threads,
                fallback_path,
                mounted_signal,
                provider,
            )
        }
        FsKind::Remote { addr: remote_addr } => {
            log::info!(
                "starting daemon with remote provider using {:?}",
                remote_addr.socket_addr()
            );
            let provider = remote_provider(remote_addr, node_creds)
                .await
                .expect("failed to create remote provider");
            FuseDaemon::new(
                config.mnt_dir,
                &config.mnt_options,
                config.threads,
                fallback_path,
                mounted_signal,
                provider,
            )
        }
    }
    .expect("failed to create FUSE daemon");

    daemon.finished().await;
}

#[tokio::main]
async fn main() {
    env_logger::init();
    let config = BtfuseConfig::new().unwrap();
    run_daemon(config, None).await;
}

#[cfg(test)]
mod test {
    use super::*;

    use btconfig::CredStoreConfig;
    use btfproto::{local_fs::ModeAuthorizer, server::new_fs_server};
    use btlib::{
        crypto::{tpm::TpmCredStore, CredStore, CredStoreMut, Creds},
        log::BuilderExt,
        Epoch, Principaled,
    };
    use btmsg::Receiver;
    use ctor::ctor;
    use std::{
        ffi::{OsStr, OsString},
        fs::Permissions,
        io::{BufRead, SeekFrom},
        net::{IpAddr, Ipv6Addr},
        num::NonZeroUsize,
        os::unix::fs::PermissionsExt,
        time::Duration,
    };
    use swtpm_harness::SwtpmHarness;
    use tempdir::TempDir;
    use tokio::{
        fs::{
            create_dir, hard_link, metadata, read, read_dir, remove_dir, remove_file, rename,
            set_permissions, write, OpenOptions, ReadDir,
        },
        io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt},
        task::JoinHandle,
    };

    /// An optional timeout to limit the time spent waiting for the FUSE daemon to start in tests.
    const TIMEOUT: Option<Duration> = Some(Duration::from_millis(1250));

    /// The log level to use when running tests.
    /// Note that the debug log level significantly reduces performance.
    const LOG_LEVEL: &str = "warn";

    #[ctor]
    fn ctor() {
        std::env::set_var("RUST_LOG", LOG_LEVEL);
        env_logger::Builder::from_default_env().btformat().init();
    }

    /// Reads `/etc/mtab` to determine if `mnt_path` is mounted. Returns true if it is and false
    /// otherwise.
    fn mounted(mnt_path: &str) -> bool {
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(false)
            .create(false)
            .open("/etc/mtab")
            .unwrap();
        let mut reader = std::io::BufReader::new(file);
        let mut line = String::with_capacity(64);
        loop {
            line.clear();
            let read = reader.read_line(&mut line).unwrap();
            if 0 == read {
                break;
            }
            let path = line.split(' ').skip(1).next().unwrap();
            if path == mnt_path {
                return true;
            }
        }
        false
    }

    /// Unmounts the file system at the given path.
    fn unmount<P: AsRef<Path>>(mnt_path: P) {
        let mnt_path = mnt_path.as_ref();
        if !mounted(mnt_path.to_str().unwrap()) {
            return;
        }
        const PROG: &str = "fusermount";
        let mnt_path = mnt_path
            .as_os_str()
            .to_str()
            .expect("failed to convert mnt_path to `str`");
        let code = std::process::Command::new(PROG)
            .args(["-z", "-u", mnt_path])
            .status()
            .expect("waiting for exit status failed")
            .code()
            .expect("code returned None");
        if code != 0 {
            panic!("{PROG} exited with a non-zero status: {code}");
        }
    }

    async fn file_names(mut read_dir: ReadDir) -> Vec<OsString> {
        let mut output = Vec::new();
        while let Some(entry) = read_dir.next_entry().await.unwrap() {
            output.push(entry.file_name());
        }
        output
    }

    const ROOT_PASSWD: &str = "password";

    struct TestCase {
        config: BtfuseConfig,
        handle: Option<JoinHandle<()>>,
        node_principal: OsString,
        stop_flag: Option<()>,
        // Note that the drop order of these fields is significant.
        _receiver: Option<Receiver>,
        _cred_store: TpmCredStore,
        _swtpm: SwtpmHarness,
        _temp_dir: TempDir,
    }

    async fn new_local() -> TestCase {
        new(false).await
    }

    async fn new_remote() -> TestCase {
        new(true).await
    }

    async fn new(remote: bool) -> TestCase {
        let tmp = TempDir::new("btfuse").unwrap();
        let (mounted_tx, mounted_rx) = oneshot::channel();
        let (swtpm, cred_store) = swtpm();
        let block_dir = tmp.path().join("bt");
        let (fs_kind, receiver) = if remote {
            let node_creds = Arc::new(cred_store.node_creds().unwrap());
            let bind_path = node_creds.bind_path().unwrap();
            block_dir.try_create_dir().unwrap();
            let local_fs = LocalFs::new_empty(block_dir, 0, node_creds.clone(), ModeAuthorizer)
                .await
                .unwrap();
            let ip_addr = IpAddr::V6(Ipv6Addr::LOCALHOST);
            let receiver = new_fs_server(ip_addr, node_creds.clone(), Arc::new(local_fs)).unwrap();
            let fs_kind = FsKind::Remote {
                addr: BlockAddr::new(ip_addr, Arc::new(bind_path)),
            };
            (fs_kind, Some(receiver))
        } else {
            (FsKind::Local { path: block_dir }, None)
        };
        let config = BtfuseConfig {
            threads: Some(NonZeroUsize::new(1).unwrap()),
            mnt_dir: tmp.path().join("mnt"),
            cred_store: CredStoreConfig::Tpm {
                path: swtpm.state_path().to_owned().into(),
                tabrmd: swtpm.tabrmd_config().to_owned(),
            },
            fs_kind,
            mnt_options: "default_permissions".to_string(),
        };
        let config_clone = config.clone();
        let handle = tokio::spawn(async move {
            run_daemon(config_clone, Some(mounted_tx)).await;
        });
        if let Some(timeout) = TIMEOUT {
            tokio::time::timeout(timeout, mounted_rx)
                .await
                .unwrap()
                .unwrap();
        } else {
            mounted_rx.await.unwrap();
        }
        let node_principal =
            OsString::from(cred_store.node_creds().unwrap().principal().to_string());
        TestCase {
            config,
            handle: Some(handle),
            node_principal,
            stop_flag: Some(()),
            _receiver: receiver,
            _temp_dir: tmp,
            _swtpm: swtpm,
            _cred_store: cred_store,
        }
    }

    fn swtpm() -> (SwtpmHarness, TpmCredStore) {
        let swtpm = SwtpmHarness::new().unwrap();
        let state_path: PathBuf = swtpm.state_path().to_owned();
        let cred_store = {
            let context = swtpm.context().unwrap();
            TpmCredStore::from_context(context, state_path.clone()).unwrap()
        };
        let root_creds = cred_store.gen_root_creds(ROOT_PASSWD).unwrap();
        let mut node_creds = cred_store.node_creds().unwrap();
        let expires = Epoch::now() + Duration::from_secs(3600);
        let writecap = root_creds
            .issue_writecap(node_creds.principal(), &mut std::iter::empty(), expires)
            .unwrap();
        cred_store
            .assign_node_writecap(&mut node_creds, writecap)
            .unwrap();
        (swtpm, cred_store)
    }

    impl TestCase {
        fn mnt_dir(&self) -> &Path {
            &self.config.mnt_dir
        }

        /// Signals to the daemon that it must stop.
        fn signal_stop(&mut self) {
            if let Some(_) = self.stop_flag.take() {
                unmount(&self.config.mnt_dir)
            }
        }

        /// Returns a future that resolves when the daemon has stopped.
        async fn stopped(&mut self) {
            if let Some(handle) = self.handle.take() {
                handle.await.expect("join failed");
            }
        }

        /// Signals to the daemon to stop and returns a future that resolves when after it has
        /// stopped.
        async fn stop(&mut self) {
            self.signal_stop();
            self.stopped().await;
        }

        fn initial_contents(&self) -> Vec<&OsStr> {
            vec![&self.node_principal]
        }
    }

    impl Drop for TestCase {
        fn drop(&mut self) {
            self.signal_stop()
        }
    }

    /// Creates a new file system and mounts it at `/tmp/btfuse.<random>/mnt` so it can be
    /// tested manually.
    //#[tokio::test]
    #[allow(dead_code)]
    async fn manual_test() {
        let mut case = new_local().await;
        case.stopped().await
    }

    async fn write_read(mut case: TestCase) -> Result<()> {
        const EXPECTED: &[u8] =
            b"The paths to failure are uncountable, yet to success there are few.";
        let file_path = case.mnt_dir().join("file");

        write(&file_path, EXPECTED).await?;

        let actual = read(&file_path).await?;
        assert_eq!(EXPECTED, actual);

        case.stop().await;
        Ok(())
    }

    #[tokio::test]
    async fn write_read_local() -> Result<()> {
        write_read(new_local().await).await
    }

    // When the current thread runtime is used the test executable does not exit after the test
    // method returns because one of the FuseDaemon blocking threads is blocked calling
    // `FuseChannel::get_request`.
    #[tokio::test(flavor = "multi_thread")]
    async fn write_read_remote() -> Result<()> {
        write_read(new_remote().await).await
    }

    async fn create_file_then_readdir(mut case: TestCase) {
        const DATA: &[u8] = b"Au revoir Shoshanna!";
        let file_name = OsStr::new("landa_dialog.txt");
        let mut expected = case.initial_contents();
        expected.push(file_name);
        let mnt_path = case.mnt_dir();
        let file_path = mnt_path.join(file_name);

        write(&file_path, DATA).await.expect("write failed");

        let first = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, first);
        let second = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, second);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_file_then_readdir_local() {
        create_file_then_readdir(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_file_then_readdir_remote() {
        create_file_then_readdir(new_remote().await).await
    }

    async fn create_then_delete_file(mut case: TestCase) {
        const DATA: &[u8] = b"The universe is hostile, so impersonal. Devour to survive";
        let file_name = OsStr::new("tool_lyrics.txt");
        let mnt_path = case.mnt_dir();
        let file_path = mnt_path.join(file_name);
        write(&file_path, DATA).await.expect("write failed");

        remove_file(&file_path).await.expect("remove_file failed");

        let expected = case.initial_contents();
        let actual = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_then_delete_file_local() {
        create_then_delete_file(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_then_delete_file_remote() {
        create_then_delete_file(new_remote().await).await
    }

    async fn hard_link_then_remove(mut case: TestCase) {
        const EXPECTED: &[u8] = b"And the lives we've reclaimed";
        let name1 = OsStr::new("refugee_lyrics.txt");
        let name2 = OsStr::new("rise_against_lyrics.txt");
        let mnt_path = case.mnt_dir();
        let path1 = mnt_path.join(name1);
        let path2 = mnt_path.join(name2);
        write(&path1, EXPECTED).await.expect("write failed");

        hard_link(&path1, &path2).await.expect("hard_link failed");
        remove_file(&path1).await.expect("remove_file failed");

        let actual = read(&path2).await.expect("read failed");
        assert_eq!(EXPECTED, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn hard_link_then_remove_local() {
        hard_link_then_remove(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn hard_link_then_remove_remote() {
        hard_link_then_remove(new_remote().await).await
    }

    async fn hard_link_then_remove_both(mut case: TestCase) {
        const EXPECTED: &[u8] = b"And the lives we've reclaimed";
        let name1 = OsStr::new("refugee_lyrics.txt");
        let name2 = OsStr::new("rise_against_lyrics.txt");
        let mnt_path = case.mnt_dir();
        let path1 = mnt_path.join(name1);
        let path2 = mnt_path.join(name2);
        write(&path1, EXPECTED).await.expect("write failed");

        hard_link(&path1, &path2).await.expect("hard_link failed");
        remove_file(&path1)
            .await
            .expect("remove_file on path1 failed");
        remove_file(&path2)
            .await
            .expect("remove_file on path2 failed");

        let expected = case.initial_contents();
        let actual = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn hard_link_then_remove_both_local() {
        hard_link_then_remove_both(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn hard_link_then_remove_both_remote() {
        hard_link_then_remove_both(new_remote().await).await
    }

    async fn set_mode_bits(mut case: TestCase) {
        const EXPECTED: u32 = libc::S_IFREG | 0o777;
        let file_path = case.mnt_dir().join("bagobits");
        write(&file_path, []).await.expect("write failed");
        let original = metadata(&file_path)
            .await
            .expect("metadata failed")
            .permissions()
            .mode();
        assert_ne!(EXPECTED, original);

        set_permissions(&file_path, Permissions::from_mode(EXPECTED))
            .await
            .expect("set_permissions failed");

        let actual = metadata(&file_path)
            .await
            .expect("metadata failed")
            .permissions()
            .mode();
        assert_eq!(EXPECTED, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn set_mode_bits_local() {
        set_mode_bits(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn set_mode_bits_remote() {
        set_mode_bits(new_remote().await).await
    }

    async fn create_directory(mut case: TestCase) {
        const EXPECTED: &str = "etc";
        let mnt_path = case.mnt_dir();
        let dir_path = mnt_path.join(EXPECTED);
        let mut expected = case.initial_contents();
        expected.push(OsStr::new(EXPECTED));

        create_dir(&dir_path).await.expect("create_dir failed");

        let actual = file_names(read_dir(mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(expected, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_directory_local() {
        create_directory(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_directory_remote() {
        create_directory(new_remote().await).await
    }

    async fn create_file_under_new_directory(mut case: TestCase) {
        const DIR_NAME: &str = "etc";
        const FILE_NAME: &str = "file";
        let mnt_path = case.mnt_dir();
        let dir_path = mnt_path.join(DIR_NAME);
        let file_path = dir_path.join(FILE_NAME);

        create_dir(&dir_path).await.expect("create_dir failed");
        write(&file_path, []).await.expect("write failed");

        let actual = file_names(read_dir(dir_path).await.expect("read_dir failed")).await;
        assert_eq!([FILE_NAME].as_slice(), actual.as_slice());

        case.stop().await;
    }

    #[tokio::test]
    async fn create_file_under_new_directory_local() {
        create_file_under_new_directory(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_file_under_new_directory_remote() {
        create_file_under_new_directory(new_remote().await).await
    }

    async fn create_then_remove_directory(mut case: TestCase) {
        const DIR_NAME: &str = "etc";
        let mnt_path = case.mnt_dir();
        let dir_path = mnt_path.join(DIR_NAME);

        create_dir(&dir_path).await.expect("create_dir failed");
        remove_dir(&dir_path).await.expect("remove_dir failed");

        let actual = file_names(read_dir(&mnt_path).await.expect("read_dir failed")).await;
        assert_eq!(case.initial_contents(), actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn create_then_remote_directory_local() {
        create_then_remove_directory(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn create_then_remote_directory_remote() {
        create_then_remove_directory(new_remote().await).await
    }

    async fn read_only_dir_cant_create_subdir(mut case: TestCase) {
        const DIR_NAME: &str = "etc";
        let dir_path = case.mnt_dir().join(DIR_NAME);
        create_dir(&dir_path).await.expect("create_dir failed");
        set_permissions(&dir_path, Permissions::from_mode(libc::S_IFDIR | 0o444))
            .await
            .expect("set_permissions failed");

        let result = create_dir(dir_path.join("sub")).await;

        let err = result.err().expect("create_dir returned `Ok`");
        let os_err = err.raw_os_error().expect("raw_os_error was empty");
        assert_eq!(os_err, libc::EACCES);

        case.stop().await;
    }

    #[tokio::test]
    async fn read_only_dir_cant_create_subdir_local() {
        read_only_dir_cant_create_subdir(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn read_only_dir_cant_create_subdir_remote() {
        read_only_dir_cant_create_subdir(new_remote().await).await
    }

    async fn read_only_dir_cant_remove_subdir(mut case: TestCase) {
        const DIR_NAME: &str = "etc";
        let dir_path = case.mnt_dir().join(DIR_NAME);
        let sub_path = dir_path.join("sub");
        create_dir(&dir_path).await.expect("create_dir failed");
        create_dir(&sub_path).await.expect("create_dir failed");
        set_permissions(&dir_path, Permissions::from_mode(libc::S_IFDIR | 0o444))
            .await
            .expect("set_permissions failed");

        let result = remove_dir(&sub_path).await;

        let err = result.err().expect("remove_dir returned `Ok`");
        let os_err = err.raw_os_error().expect("raw_os_error was empty");
        assert_eq!(os_err, libc::EACCES);

        case.stop().await;
    }

    #[tokio::test]
    async fn read_only_dir_cant_remove_subdir_local() {
        read_only_dir_cant_remove_subdir(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn read_only_dir_cant_remove_subdir_remote() {
        read_only_dir_cant_remove_subdir(new_remote().await).await
    }

    async fn rename_file(mut case: TestCase) {
        const FILE_NAME: &str = "parabola.txt";
        const EXPECTED: &[u8] = b"We are eternal all this pain is an illusion";
        let src_path = case.mnt_dir().join(FILE_NAME);
        let dst_path = case.mnt_dir().join("parabola_lyrics.txt");

        write(&src_path, EXPECTED).await.unwrap();
        rename(&src_path, &dst_path).await.unwrap();

        let actual = read(&dst_path).await.unwrap();
        assert_eq!(EXPECTED, actual);

        case.stop().await;
    }

    #[tokio::test]
    async fn rename_file_local() {
        rename_file(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn rename_file_remote() {
        rename_file(new_remote().await).await
    }

    async fn write_read_with_file_struct(mut case: TestCase) {
        const FILE_NAME: &str = "big.dat";
        const LEN: usize = btlib::SECTOR_SZ_DEFAULT + 1;
        fn fill(buf: &mut Vec<u8>, value: u8) {
            buf.clear();
            buf.extend(std::iter::repeat(value).take(buf.capacity()));
        }

        let file_path = case.mnt_dir().join(FILE_NAME);
        let mut buf = vec![1u8; LEN];
        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(&file_path)
            .await
            .unwrap();

        file.write_all(&buf).await.unwrap();
        fill(&mut buf, 2);
        file.write_all(&buf).await.unwrap();
        file.rewind().await.unwrap();
        let mut actual = vec![0u8; LEN];
        file.read_exact(&mut actual).await.unwrap();

        fill(&mut buf, 1);
        assert_eq!(buf, actual);

        drop(file);
        case.stop().await;
    }

    #[tokio::test]
    async fn write_read_with_file_struct_local() {
        write_read_with_file_struct(new_local().await).await
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn write_read_with_file_struct_remote() {
        write_read_with_file_struct(new_remote().await).await
    }

    /// KMC: This test is currently not working, and I've not been able to figure out why, nor
    /// reproduce it at a lower layer of the stack.
    async fn read_more_than_whats_buffered(mut case: TestCase) {
        const FILE_NAME: &str = "big.dat";
        const SECT_SZ: usize = btlib::SECTOR_SZ_DEFAULT;
        const DIVISOR: usize = 8;
        const READ_SZ: usize = SECT_SZ / DIVISOR;

        let file_path = case.mnt_dir().join(FILE_NAME);
        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .open(&file_path)
            .await
            .unwrap();

        let mut buf = vec![1u8; 2 * SECT_SZ];
        file.write_all(&buf).await.unwrap();
        file.flush().await.unwrap();
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&file_path)
            .await
            .unwrap();
        file.seek(SeekFrom::Start(SECT_SZ as u64)).await.unwrap();
        let mut actual = vec![0u8; READ_SZ];
        file.read_exact(&mut actual).await.unwrap();

        buf.truncate(READ_SZ);
        assert!(buf == actual);

        case.stop().await;
    }

    //#[tokio::test]
    #[allow(dead_code)]
    async fn read_more_than_whats_buffered_local() {
        read_more_than_whats_buffered(new_local().await).await
    }

    //#[tokio::test(flavor = "multi_thread")]
    #[allow(dead_code)]
    async fn read_more_than_whats_buffered_remote() {
        read_more_than_whats_buffered(new_remote().await).await
    }
}

#[cfg(test)]
mod config_tests {
    use super::{BtfuseConfig, FsKind};

    use std::{net::IpAddr, num::NonZeroUsize, path::PathBuf, sync::Arc};

    use btconfig::CredStoreConfig;
    use btlib::BlockPath;
    use btmsg::BlockAddr;
    use figment::Jail;

    #[test]
    fn fs_kind_local() {
        Jail::expect_with(|jail| {
            const EXPECTED_PATH: &str = "/tmp/blocks";
            let expected = FsKind::Local {
                path: EXPECTED_PATH.into(),
            };
            jail.set_env("BT_FSKIND_TYPE", "Local");
            jail.set_env("BT_FSKIND_PATH", EXPECTED_PATH);

            let config = BtfuseConfig::new().unwrap();

            assert_eq!(expected, config.fs_kind);

            Ok(())
        })
    }

    #[test]
    fn fs_kind_remote() {
        Jail::expect_with(|jail| {
            let expected_ip = IpAddr::from([127, 0, 0, 42]);
            let expected_path = Arc::new(BlockPath::try_from(
                "/0!zX_LMUVQO2Y7mgDomQB8ZdNsXKlykpHs-zPX9C3ztII/0!vVB5rOb3NFjzaZl_wlH3jqhBaYV7uuxrk3_s42xLnzg"
            ).unwrap());
            let expected_addr = BlockAddr::new(expected_ip, expected_path.clone());
            let expected = FsKind::Remote {
                addr: expected_addr,
            };
            jail.set_env("BT_FSKIND_TYPE", "Remote");
            jail.set_env("BT_FSKIND_ADDR_IPADDR", expected_ip);
            jail.set_env("BT_FSKIND_ADDR_PATH", expected_path.as_ref());

            let config = BtfuseConfig::new().unwrap();

            assert_eq!(expected, config.fs_kind);

            Ok(())
        })
    }

    #[test]
    fn mnt_dir() {
        Jail::expect_with(|jail| {
            let expected = PathBuf::from("/tmp/btfuse_mnt");
            jail.set_env("BT_MNTDIR", expected.display());

            let config = BtfuseConfig::new().unwrap();

            assert_eq!(expected, config.mnt_dir);

            Ok(())
        })
    }

    #[test]
    fn mnt_options() {
        Jail::expect_with(|jail| {
            let expected = "default_permissions";
            jail.set_env("BT_MNTOPTIONS", expected);

            let config = BtfuseConfig::new().unwrap();

            assert_eq!(expected, &config.mnt_options);

            Ok(())
        })
    }

    #[test]
    fn threads_is_set() {
        Jail::expect_with(|jail| {
            let expected = Some(NonZeroUsize::new(8).unwrap());
            jail.set_env("BT_THREADS", expected.unwrap().get());

            let config = BtfuseConfig::new().unwrap();

            assert_eq!(expected, config.threads);

            Ok(())
        })
    }

    #[test]
    fn threads_is_not_set() {
        Jail::expect_with(|_jail| {
            let expected = None;

            let config = BtfuseConfig::new().unwrap();

            assert_eq!(expected, config.threads);

            Ok(())
        })
    }

    #[test]
    fn cred_store_path() {
        Jail::expect_with(|jail| {
            let expected = PathBuf::from("/tmp/secrets/file_credstore");
            jail.set_env("BT_CREDSTORE_TYPE", "File");
            jail.set_env("BT_CREDSTORE_PATH", expected.display());

            let config = BtfuseConfig::new().unwrap();

            let success = if let CredStoreConfig::File { path: actual } = config.cred_store {
                expected == actual
            } else {
                false
            };
            assert!(success);

            Ok(())
        })
    }
}