Blocktree/crates/btfuse/src/fuse_fs.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
use btfproto::{msg::*, server::FsProvider};
use btlib::{
    bterr, collections::Bijection, error::DisplayErr, AuthzAttrs, BlockId, BlockMetaSecrets,
    BlockPath, Epoch, Result,
};
use btserde::read_from;
use core::{ffi::CStr, future::Future, time::Duration};
use fuse_backend_rs::{
    abi::fuse_abi::{stat64, Attr, CreateIn},
    api::filesystem::{
        Context, DirEntry, Entry, FileSystem, FsOptions, OpenOptions, SetattrValid, ZeroCopyReader,
        ZeroCopyWriter,
    },
};
use log::{debug, error};
use std::{
    io::{self, Result as IoResult},
    sync::{Arc, RwLock},
};
use tokio::runtime::Handle;

pub use private::FuseFs;

mod private {
    use super::*;

    trait BlockMetaSecretsExt {
        fn attr(&self) -> Result<Attr>;

        fn stat(&self) -> Result<stat64> {
            self.attr().map(|e| e.into())
        }
    }

    impl BlockMetaSecretsExt for BlockMetaSecrets {
        fn attr(&self) -> Result<Attr> {
            Ok(Attr {
                ino: self.block_id.inode,
                size: self.size,
                atime: self.atime.value(),
                mtime: self.mtime.value(),
                ctime: self.ctime.value(),
                atimensec: 0,
                mtimensec: 0,
                ctimensec: 0,
                mode: self.mode,
                nlink: self.nlink,
                uid: self.uid,
                gid: self.gid,
                rdev: 0,
                blksize: self.sect_sz.try_into().map_err(|_| {
                    bterr!("BlockMetaSecrets::sect_sz could not be converted to a u32")
                })?,
                blocks: self.sectors(),
                flags: 0,
            })
        }
    }

    fn block_on<F: Future>(future: F) -> F::Output {
        Handle::current().block_on(future)
    }

    pub struct FuseFs<P> {
        provider: P,
        path_by_luid: Bijection<u32, Arc<BlockPath>>,
        ruid_by_path: RwLock<Bijection<Arc<BlockPath>, u32>>,
    }

    impl<P> FuseFs<P> {
        pub fn new(provider: P, fallback_path: Arc<BlockPath>) -> Self {
            let proc_uid = unsafe { libc::geteuid() };
            Self {
                provider,
                /// luid: Local UID
                path_by_luid: Bijection::new(proc_uid, fallback_path.clone()),
                // ruid: Remote UID
                ruid_by_path: RwLock::new(Bijection::new(fallback_path, 0)),
            }
        }
        fn path_from_luid(&self, luid: u32) -> &Arc<BlockPath> {
            self.path_by_luid.value(&luid)
        }

        fn luid_from_ruid(&self, ruid: u32) -> Result<u32> {
            let guard = self.ruid_by_path.read().display_err()?;
            let path = guard.key(&ruid);
            Ok(*self.path_by_luid.key(path))
        }

        fn convert_ruid(&self, mut attrs: BlockMetaSecrets) -> Result<BlockMetaSecrets> {
            attrs.uid = self.luid_from_ruid(attrs.uid)?;
            Ok(attrs)
        }

        fn fuse_entry(&self, attrs: btfproto::msg::Entry) -> Result<Entry> {
            let btfproto::msg::Entry {
                attr,
                attr_timeout,
                entry_timeout,
            } = attrs;
            let attr = self.convert_ruid(attr)?;
            let BlockId { inode, generation } = attr.block_id;
            let attr = attr.stat()?;
            Ok(Entry {
                inode,
                generation,
                attr,
                attr_flags: 0,
                attr_timeout,
                entry_timeout,
            })
        }
    }

    impl<P: 'static + FsProvider> FuseFs<P> {
        async fn load_authz_attrs(
            provider: &P,
            from: &Arc<BlockPath>,
            path: &BlockPath,
        ) -> Result<AuthzAttrs> {
            let mut parent = SpecInodes::RootDir.into();
            for component in path.components() {
                let msg = Lookup {
                    parent,
                    name: component,
                };
                let LookupReply { inode, .. } = provider.lookup(from, msg).await?;
                parent = inode;
            }
            let msg = Open {
                inode: parent,
                flags: FlagValue::ReadOnly.into(),
            };
            let OpenReply { handle, .. } = provider.open(from, msg).await?;
            let msg = Read {
                inode: parent,
                handle,
                offset: 0,
                size: u64::MAX,
            };
            let guard = provider.read(from, msg).await?;
            let mut slice: &[u8] = &guard;
            read_from(&mut slice).map_err(|err| err.into())
        }
    }

    unsafe impl<P: Sync> Sync for FuseFs<P> {}

    impl<P: 'static + FsProvider> FileSystem for FuseFs<P> {
        type Inode = btfproto::Inode;
        type Handle = btfproto::Handle;

        fn init(&self, _capable: FsOptions) -> io::Result<FsOptions> {
            log::debug!("init called");
            let provider_ref = &self.provider;
            let default_path = self.path_by_luid.k2v().default();
            let default_path_clone = default_path.clone();
            let attrs = block_on(async move {
                Self::load_authz_attrs(
                    provider_ref,
                    &default_path_clone,
                    default_path_clone.as_ref(),
                )
                .await
            })?;
            let mut guard = self.ruid_by_path.write().display_err()?;
            guard.insert(default_path.clone(), attrs.uid);
            Ok(FsOptions::empty())
        }

        fn lookup(&self, ctx: &Context, parent: Self::Inode, name: &CStr) -> IoResult<Entry> {
            log::debug!("lookup called");
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let name = name.to_str().display_err()?;
                let msg = Lookup { parent, name };
                let entry = match self.provider.lookup(path, msg).await {
                    Ok(LookupReply { entry, .. }) => entry,
                    Err(err) => {
                        return match err.downcast::<io::Error>() {
                            Ok(err) => {
                                debug!("lookup returned io::Error: {err}");
                                Err(err)
                            }
                            Err(err) => {
                                debug!("lookup returned an unknown error: {err}");
                                Err(io::Error::new(io::ErrorKind::Other, err.to_string()))
                            }
                        }
                    }
                };
                let entry = self.fuse_entry(entry)?;
                Ok(entry)
            })
        }

        fn forget(&self, ctx: &Context, inode: Self::Inode, count: u64) {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = Forget { inode, count };
                if let Err(err) = self.provider.forget(path, msg).await {
                    error!("error when sending the Forget message: {err}");
                }
            })
        }

        fn create(
            &self,
            ctx: &Context,
            parent: Self::Inode,
            name: &CStr,
            args: CreateIn,
        ) -> IoResult<(Entry, Option<Self::Handle>, OpenOptions)> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let name = name.to_str().display_err()?;
                let msg = Create {
                    parent,
                    name,
                    flags: Flags::new(args.flags as i32),
                    mode: args.mode,
                    umask: args.umask,
                };
                let CreateReply { entry, handle, .. } = self.provider.create(path, msg).await?;
                let entry = self.fuse_entry(entry)?;
                Ok((entry, Some(handle), OpenOptions::empty()))
            })
        }

        fn mkdir(
            &self,
            ctx: &Context,
            parent: Self::Inode,
            name: &CStr,
            mode: u32,
            umask: u32,
        ) -> io::Result<Entry> {
            let args = CreateIn {
                flags: FlagValue::Directory.value() as u32,
                mode,
                umask,
                fuse_flags: 0,
            };
            let (entry, ..) = self.create(ctx, parent, name, args)?;
            Ok(entry)
        }

        fn open(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            flags: u32,
            _fuse_flags: u32,
        ) -> IoResult<(Option<Self::Handle>, OpenOptions)> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = Open {
                    inode,
                    flags: Flags::new(flags as i32),
                };
                let handle = match self.provider.open(path, msg).await {
                    Ok(OpenReply { handle, .. }) => handle,
                    Err(err) => {
                        error!("FsProvider::open returned an error: {err}");
                        return Err(err.into());
                    }
                };
                Ok((Some(handle), OpenOptions::empty()))
            })
        }

        fn release(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            flags: u32,
            handle: Self::Handle,
            _flush: bool,
            _flock_release: bool,
            _lock_owner: Option<u64>,
        ) -> io::Result<()> {
            self.releasedir(ctx, inode, flags, handle)
        }

        fn opendir(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            flags: u32,
        ) -> IoResult<(Option<Self::Handle>, OpenOptions)> {
            let flags = flags | libc::O_DIRECTORY as u32;
            self.open(ctx, inode, flags, 0)
        }

        fn releasedir(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            _flags: u32,
            handle: Self::Handle,
        ) -> io::Result<()> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = Close { inode, handle };
                self.provider.close(path, msg).await?;
                Ok(())
            })
        }

        fn read(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            w: &mut dyn ZeroCopyWriter,
            size: u32,
            mut offset: u64,
            _lock_owner: Option<u64>,
            _flags: u32,
        ) -> IoResult<usize> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let total_size: usize = size.try_into().display_err()?;
                let mut total_written = 0;
                while total_written < total_size {
                    let msg = Read {
                        inode,
                        handle,
                        offset,
                        size: size as u64,
                    };
                    let guard = self.provider.read(path, msg).await?;
                    let slice: &[u8] = &guard;
                    if slice.is_empty() {
                        break;
                    }
                    w.write_all(&guard)?;
                    total_written += slice.len();
                    let len: u64 = slice.len().try_into().display_err()?;
                    offset += len;
                }
                Ok(total_written)
            })
        }

        fn write(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            r: &mut dyn ZeroCopyReader,
            size: u32,
            offset: u64,
            _lock_owner: Option<u64>,
            _delayed_write: bool,
            _flags: u32,
            _fuse_flags: u32,
        ) -> IoResult<usize> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let size: usize = size.try_into().display_err()?;
                // TODO: Eliminate this copying, or at least use a pool of buffers to avoid
                // allocating on every write. We could pass `r` to the provider if it were Send.
                let mut buf = Vec::with_capacity(size);
                r.read_to_end(&mut buf)?;
                let msg = Write {
                    inode,
                    handle,
                    offset,
                    data: buf.as_slice(),
                };
                let WriteReply { written, .. } = self.provider.write(path, msg).await?;
                Ok(written.try_into().display_err()?)
            })
        }

        fn flush(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            _lock_owner: u64,
        ) -> io::Result<()> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = Flush { inode, handle };
                self.provider.flush(path, msg).await?;
                Ok(())
            })
        }

        fn readdir(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            size: u32,
            offset: u64,
            add_entry: &mut dyn FnMut(DirEntry) -> io::Result<usize>,
        ) -> io::Result<()> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = ReadDir {
                    inode,
                    handle,
                    limit: 0,
                    state: offset,
                };
                let ReadDirReply { entries, .. } = self.provider.read_dir(path, msg).await?;
                let mut size: usize = size.try_into().display_err()?;
                for (index, (name, entry)) in entries.into_iter().enumerate() {
                    // We do not expose non-standard directory entries via FUSE.
                    if entry.kind() == libc::DT_UNKNOWN {
                        continue;
                    }
                    let inode = entry.inode();
                    let offset = (index as u64) + 1;
                    let dir_entry = DirEntry {
                        ino: inode,
                        offset,
                        type_: entry.kind() as u32,
                        name: name.as_bytes(),
                    };
                    size = size.saturating_sub(add_entry(dir_entry)?);
                    if size == 0 {
                        break;
                    }
                }
                Ok(())
            })
        }

        fn link(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            new_parent: Self::Inode,
            new_name: &CStr,
        ) -> io::Result<Entry> {
            debug!("link called");
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let name = new_name.to_str().display_err()?;
                let msg = Link {
                    inode,
                    new_parent,
                    name,
                };
                let LinkReply { entry, .. } = self.provider.link(path, msg).await?;
                let entry = self.fuse_entry(entry)?;
                Ok(entry)
            })
        }

        fn unlink(&self, ctx: &Context, parent: Self::Inode, name: &CStr) -> io::Result<()> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let name = name.to_str().display_err()?;
                let msg = Unlink { parent, name };
                self.provider.unlink(path, msg).await?;
                Ok(())
            })
        }

        fn rmdir(&self, ctx: &Context, parent: Self::Inode, name: &CStr) -> io::Result<()> {
            self.unlink(ctx, parent, name)
        }

        fn rename(
            &self,
            ctx: &Context,
            olddir: Self::Inode,
            oldname: &CStr,
            newdir: Self::Inode,
            newname: &CStr,
            _flags: u32,
        ) -> io::Result<()> {
            let Entry { inode, .. } = self.lookup(ctx, olddir, oldname)?;
            let result = (move || {
                self.link(ctx, inode, newdir, newname)?;
                self.unlink(ctx, olddir, oldname)?;
                Ok(())
            })();
            self.forget(ctx, inode, 1);
            result
        }

        fn getattr(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            handle: Option<Self::Handle>,
        ) -> IoResult<(stat64, Duration)> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = ReadMeta { inode, handle };
                let ReadMetaReply {
                    attrs, valid_for, ..
                } = self.provider.read_meta(path, msg).await?;
                let attrs = self.convert_ruid(attrs)?;
                let stat = attrs.stat()?;
                Ok((stat, valid_for))
            })
        }

        fn setattr(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            attr: stat64,
            handle: Option<Self::Handle>,
            valid: SetattrValid,
        ) -> IoResult<(stat64, Duration)> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let mut msg_attrs = Attrs::default();
                let mut attrs_set = AttrsSet::none();
                if valid.intersects(SetattrValid::MODE) {
                    msg_attrs.mode = attr.st_mode;
                    attrs_set |= AttrsSet::MODE;
                }
                if valid.intersects(SetattrValid::UID) {
                    msg_attrs.uid = attr.st_uid;
                    attrs_set |= AttrsSet::UID;
                }
                if valid.intersects(SetattrValid::GID) {
                    msg_attrs.gid = attr.st_gid;
                    attrs_set |= AttrsSet::GID;
                }
                if valid.intersects(SetattrValid::ATIME) {
                    let atime: u64 = attr.st_atime.try_into().display_err()?;
                    msg_attrs.atime = Epoch::from(atime);
                    attrs_set |= AttrsSet::ATIME;
                }
                if valid.intersects(SetattrValid::MTIME) {
                    let mtime: u64 = attr.st_mtime.try_into().display_err()?;
                    msg_attrs.mtime = Epoch::from(mtime);
                    attrs_set |= AttrsSet::MTIME;
                }
                if valid.intersects(SetattrValid::CTIME) {
                    let ctime: u64 = attr.st_ctime.try_into().display_err()?;
                    msg_attrs.ctime = Epoch::from(ctime);
                    attrs_set |= AttrsSet::CTIME;
                }
                let msg = WriteMeta {
                    inode,
                    handle,
                    attrs: msg_attrs,
                    attrs_set,
                };
                let WriteMetaReply {
                    attrs, valid_for, ..
                } = self.provider.write_meta(path, msg).await?;
                Ok((attrs.stat()?, valid_for))
            })
        }

        fn fsync(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            _datasync: bool,
            handle: Self::Handle,
        ) -> IoResult<()> {
            block_on(async move {
                let path = self.path_from_luid(ctx.uid);
                let msg = Flush { inode, handle };
                self.provider.flush(path, msg).await?;
                Ok(())
            })
        }

        fn fsyncdir(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            datasync: bool,
            handle: Self::Handle,
        ) -> IoResult<()> {
            self.fsync(ctx, inode, datasync, handle)
        }

        fn fallocate(
            &self,
            ctx: &Context,
            inode: Self::Inode,
            handle: Self::Handle,
            mode: u32,
            offset: u64,
            length: u64,
        ) -> IoResult<()> {
            block_on(async move {
                if mode != 0 {
                    error!("a non-zero mode argument was given to async_fallocate: {mode}");
                    return Err(io::Error::from_raw_os_error(libc::ENOTSUP));
                }
                let path = self.path_from_luid(ctx.uid);
                let msg = Allocate {
                    inode,
                    handle,
                    offset: Some(offset),
                    size: length,
                };
                self.provider.allocate(path, msg).await?;
                Ok(())
            })
        }
    }
}

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

    use btfproto::Inode;
    use btfproto_tests::local_fs_tests::{ConcreteFs, LocalFsTest};
    use fuse_backend_rs::api::filesystem::Context;
    use std::ffi::CString;
    use tempdir::TempDir;

    struct FuseFsTest {
        _dir: TempDir,
        fs: FuseFs<ConcreteFs>,
    }

    impl FuseFsTest {
        async fn new_empty() -> Self {
            let case = LocalFsTest::new_empty().await;
            let from = case.from().to_owned();
            let (dir, fs, ..) = case.into_parts();
            let fs = FuseFs::new(fs, from);
            Self { _dir: dir, fs }
        }

        fn fs(&self) -> &FuseFs<ConcreteFs> {
            &self.fs
        }

        fn ctx(&self) -> Context {
            Context {
                uid: 0,
                gid: 0,
                pid: 0,
            }
        }
    }

    #[tokio::test]
    async fn lookup_file_exists() {
        let case = FuseFsTest::new_empty().await;
        tokio::task::spawn_blocking(move || {
            let fuse_fs = case.fs();
            let root: Inode = SpecInodes::RootDir.into();
            let ctx = case.ctx();
            let name = CString::new("file.txt").unwrap();

            let args = CreateIn {
                flags: libc::O_RDWR as u32,
                mode: 0o644,
                umask: 0,
                fuse_flags: 0,
            };
            let (expected, ..) = fuse_fs.create(&ctx, root, &name, args).unwrap();
            let actual = fuse_fs.lookup(&ctx, root, &name).unwrap();

            assert_eq!(expected.inode, actual.inode);
            assert_eq!(expected.generation, actual.generation);
            assert_eq!(expected.attr, actual.attr);
            assert_eq!(expected.attr_flags, actual.attr_flags);
            assert_eq!(expected.attr_timeout, actual.attr_timeout);
            assert_eq!(expected.entry_timeout, actual.entry_timeout);
        })
        .await
        .unwrap();
    }

    #[tokio::test]
    async fn setattr() {
        macro_rules! check {
            (
                $actual:ident,
                $entry:ident,
                $getattr_return:ident,
                $setattr_return:ident,
                $field:ident
            ) => {
                assert_ne!($actual.$field, $entry.attr.$field);
                assert_eq!($actual.$field, $setattr_return.$field);
                assert_eq!($actual.$field, $getattr_return.$field);
            };
        }

        let case = FuseFsTest::new_empty().await;
        tokio::task::spawn_blocking(move || {
            let fuse_fs = case.fs();
            let root: Inode = SpecInodes::RootDir.into();
            let ctx = case.ctx();
            let name = CString::new("file.txt").unwrap();

            let args = CreateIn {
                flags: libc::O_RDWR as u32,
                mode: 0o644,
                umask: 0,
                fuse_flags: 0,
            };
            let (entry, handle, ..) = fuse_fs.create(&ctx, root, &name, args).unwrap();
            let inode = entry.inode;
            let actual = {
                let mut actual = stat64::from(Attr::default());
                actual.st_mode = 0o777;
                actual.st_atime = 21323;
                actual.st_mtime = 21290;
                actual.st_ctime = 119200;
                actual
            };
            let valid = SetattrValid::MODE
                | SetattrValid::UID
                | SetattrValid::GID
                | SetattrValid::ATIME
                | SetattrValid::MTIME
                | SetattrValid::CTIME;
            let (setattr_return, ..) = fuse_fs.setattr(&ctx, inode, actual, handle, valid).unwrap();
            let (getattr_return, ..) = fuse_fs.getattr(&ctx, inode, handle).unwrap();

            check!(actual, entry, getattr_return, setattr_return, st_mode);
            check!(actual, entry, getattr_return, setattr_return, st_atime);
            check!(actual, entry, getattr_return, setattr_return, st_mtime);
            check!(actual, entry, getattr_return, setattr_return, st_ctime);
        })
        .await
        .unwrap();
    }
}