Server implemented. Linux client implemented.

furumi-server/src/main.rs

@@ -0,0 +1,68 @@
+pub mod vfs;
+pub mod security;
+pub mod server;
+
+use std::path::PathBuf;
+use std::sync::Arc;
+use clap::Parser;
+use tonic::transport::Server;
+use vfs::local::LocalVfs;
+use furumi_common::proto::remote_file_system_server::RemoteFileSystemServer;
+use server::RemoteFileSystemImpl;
+use security::AuthInterceptor;
+
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct Args {
+    /// IP address and port to bind the server to
+    #[arg(short, long, env = "FURUMI_BIND", default_value = "[::1]:50051")]
+    bind: String,
+
+    /// Document root directory to expose via VFS
+    #[arg(short, long, env = "FURUMI_ROOT", default_value = ".")]
+    root: PathBuf,
+
+    /// Authentication Bearer token (leave empty to disable auth)
+    #[arg(short, long, env = "FURUMI_TOKEN", default_value = "")]
+    token: String,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing_subscriber::fmt::init();
+
+    let args = Args::parse();
+    let addr = args.bind.parse()?;
+    
+    // Resolve the document root to an absolute path for safety and clarity
+    let root_path = std::fs::canonicalize(&args.root)
+        .unwrap_or_else(|_| args.root.clone());
+    
+    if !root_path.exists() || !root_path.is_dir() {
+        eprintln!("Error: Root path {:?} does not exist or is not a directory", root_path);
+        std::process::exit(1);
+    }
+
+    let vfs = Arc::new(LocalVfs::new(&root_path));
+    let remote_fs = RemoteFileSystemImpl::new(vfs);
+    let auth = AuthInterceptor::new(args.token.clone());
+    let svc = RemoteFileSystemServer::with_interceptor(remote_fs, auth.clone());
+
+    println!("Furumi-ng Server listening on {}", addr);
+    if args.token.is_empty() {
+        println!("WARNING: Authentication is DISABLED");
+    } else {
+        println!("Authentication is enabled (Bearer token required)");
+    }
+    println!("Document Root: {:?}", root_path);
+
+    Server::builder()
+        // Enable TCP Keep-Alive and HTTP2 Ping to keep connections alive for long media streams
+        .tcp_keepalive(Some(std::time::Duration::from_secs(60)))
+        .http2_keepalive_interval(Some(std::time::Duration::from_secs(60)))
+        .add_service(svc)
+        .serve(addr)
+        .await?;
+
+    Ok(())
+}

furumi-server/src/security/mod.rs

@@ -0,0 +1,83 @@
+use std::path::{Component, Path};
+use tonic::{Request, Status};
+
+/// Sanitizes a path strictly to prevent Path Traversal.
+/// Rejects paths containing absolute routes outside the conceptual root
+/// or any parent directory (`..`) traversal components.
+pub fn sanitize_path(input: &str) -> std::result::Result<String, Status> {
+    let path = Path::new(input);
+    let mut normalized = std::path::PathBuf::new();
+
+    for component in path.components() {
+        match component {
+            Component::ParentDir => {
+                if !normalized.pop() {
+                    return Err(Status::permission_denied("Path traversal attempt"));
+                }
+            }
+            Component::Normal(c) => normalized.push(c),
+            Component::CurDir | Component::RootDir | Component::Prefix(_) => {
+                // Ignore RootDir (making absolute paths logic relative to VFS root)
+                // Ignore Prefix (Windows C:)
+                // Ignore CurrentDir (.)
+            }
+        }
+    }
+
+    Ok(normalized.to_string_lossy().into_owned())
+}
+
+/// A simple tonic interceptor for token-based authorization.
+/// We expect a `authorization: Bearer <token>` header.
+#[derive(Clone)]
+pub struct AuthInterceptor {
+    expected_token: String,
+}
+
+impl AuthInterceptor {
+    pub fn new(token: String) -> Self {
+        Self { expected_token: token }
+    }
+}
+
+impl tonic::service::Interceptor for AuthInterceptor {
+    fn call(&mut self, req: Request<()>) -> std::result::Result<Request<()>, Status> {
+        // If no token is configured on the server, allow all (or reject all based on policy).
+        // Here we assume if expected_token is empty, auth is disabled.
+        if self.expected_token.is_empty() {
+            return Ok(req);
+        }
+
+        match req.metadata().get("authorization") {
+            Some(t) => {
+                let token_str = t.to_str().unwrap_or("");
+                let expected_header = format!("Bearer {}", self.expected_token);
+                if token_str == expected_header {
+                    Ok(req)
+                } else {
+                    Err(Status::unauthenticated("Invalid token"))
+                }
+            }
+            _ => Err(Status::unauthenticated("Missing authorization header")),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_sanitize_path_normal() {
+        assert_eq!(sanitize_path("etc/passwd").unwrap(), "etc/passwd");
+        assert_eq!(sanitize_path("/etc/passwd").unwrap(), "etc/passwd");
+        assert_eq!(sanitize_path("foo/./bar").unwrap(), "foo/bar");
+    }
+
+    #[test]
+    fn test_sanitize_path_traversal() {
+        assert!(sanitize_path("../etc/passwd").is_err());
+        assert!(sanitize_path("/../etc/passwd").is_err());
+        assert!(sanitize_path("foo/../../etc/passwd").is_err());
+    }
+}

furumi-server/src/server.rs

@@ -0,0 +1,101 @@
+use std::pin::Pin;
+use tokio_stream::Stream;
+use tonic::{Request, Response, Status};
+
+use crate::vfs::VirtualFileSystem;
+use furumi_common::proto::{
+    remote_file_system_server::RemoteFileSystem, AttrResponse, DirEntry, FileChunk,
+    PathRequest, ReadRequest,
+};
+use crate::security::sanitize_path;
+
+pub struct RemoteFileSystemImpl<V: VirtualFileSystem> {
+    vfs: std::sync::Arc<V>,
+}
+
+impl<V: VirtualFileSystem> RemoteFileSystemImpl<V> {
+    pub fn new(vfs: std::sync::Arc<V>) -> Self {
+        Self { vfs }
+    }
+}
+
+#[tonic::async_trait]
+impl<V: VirtualFileSystem> RemoteFileSystem for RemoteFileSystemImpl<V> {
+    async fn get_attr(
+        &self,
+        request: Request<PathRequest>,
+    ) -> Result<Response<AttrResponse>, Status> {
+        let req = request.into_inner();
+        let safe_path = sanitize_path(&req.path)?;
+
+        match self.vfs.get_attr(&safe_path).await {
+            Ok(attr) => Ok(Response::new(attr)),
+            Err(e) => Err(Status::internal(e.to_string())),
+        }
+    }
+
+    type ReadDirStream = Pin<
+        Box<
+            dyn Stream<Item = Result<DirEntry, Status>> + Send + 'static,
+        >,
+    >;
+
+    async fn read_dir(
+        &self,
+        request: Request<PathRequest>,
+    ) -> Result<Response<Self::ReadDirStream>, Status> {
+        let req = request.into_inner();
+        let safe_path = sanitize_path(&req.path)?;
+
+        match self.vfs.read_dir(&safe_path).await {
+            Ok(mut rx) => {
+                let stream = async_stream::try_stream! {
+                    while let Some(result) = rx.recv().await {
+                        match result {
+                            Ok(entry) => yield entry,
+                            Err(e) => Err(Status::internal(e.to_string()))?,
+                        }
+                    }
+                };
+                Ok(Response::new(Box::pin(stream) as Self::ReadDirStream))
+            }
+            Err(e) => Err(Status::internal(e.to_string())),
+        }
+    }
+
+    type ReadFileStream = Pin<
+        Box<
+            dyn Stream<Item = Result<FileChunk, Status>> + Send + 'static,
+        >,
+    >;
+
+    async fn read_file(
+        &self,
+        request: Request<ReadRequest>,
+    ) -> Result<Response<Self::ReadFileStream>, Status> {
+        let req = request.into_inner();
+        let safe_path = sanitize_path(&req.path)?;
+
+        let sanitized_req = ReadRequest {
+            path: safe_path,
+            offset: req.offset,
+            size: req.size,
+            chunk_size: req.chunk_size,
+        };
+
+        match self.vfs.read_file(sanitized_req).await {
+            Ok(mut rx) => {
+                let stream = async_stream::try_stream! {
+                    while let Some(result) = rx.recv().await {
+                        match result {
+                            Ok(chunk) => yield chunk,
+                            Err(e) => Err(Status::internal(e.to_string()))?,
+                        }
+                    }
+                };
+                Ok(Response::new(Box::pin(stream) as Self::ReadFileStream))
+            }
+            Err(e) => Err(Status::internal(e.to_string())),
+        }
+    }
+}

furumi-server/src/vfs/local.rs

@@ -0,0 +1,197 @@
+use super::VirtualFileSystem;
+use furumi_common::proto::{AttrResponse, DirEntry, FileChunk, ReadRequest};
+use anyhow::{Context, Result};
+use std::os::unix::fs::MetadataExt;
+use std::path::{Path, PathBuf};
+use tokio::fs;
+use tokio::io::{AsyncReadExt, AsyncSeekExt};
+use tokio::sync::mpsc::{self, Receiver};
+
+pub struct LocalVfs {
+    root: PathBuf,
+}
+
+impl LocalVfs {
+    pub fn new<P: AsRef<Path>>(root: P) -> Self {
+        Self {
+            root: root.as_ref().to_path_buf(),
+        }
+    }
+
+    /// Appends the sanitized relative path to the root.
+    fn resolve_path(&self, relative_path: &str) -> Result<PathBuf> {
+        let rel_path = Path::new(relative_path);
+        // Note: Security module will handle path traversal sanitization before reaching this point.
+        // We assume `relative_path` is safe and strictly logical here.
+        let stripped = rel_path.strip_prefix("/").unwrap_or(rel_path);
+        Ok(self.root.join(stripped))
+    }
+}
+
+#[async_trait::async_trait]
+impl VirtualFileSystem for LocalVfs {
+    async fn get_attr(&self, path: &str) -> Result<AttrResponse> {
+        let full_path = self.resolve_path(path)?;
+        let metadata = fs::metadata(&full_path).await.context("Failed to get metadata")?;
+
+        Ok(AttrResponse {
+            size: metadata.len(),
+            mode: metadata.mode(),
+            mtime: metadata.mtime() as u64,
+        })
+    }
+
+    async fn read_dir(&self, path: &str) -> Result<Receiver<Result<DirEntry, anyhow::Error>>> {
+        let full_path = self.resolve_path(path)?;
+        let mut dir = fs::read_dir(&full_path).await.context("Failed to read directory")?;
+
+        let (tx, rx) = mpsc::channel(100);
+
+        tokio::spawn(async move {
+            while let Ok(Some(entry)) = dir.next_entry().await {
+                // In Linux, DT_REG = 8, DT_DIR = 4. We can approximate this.
+                let file_type = entry.file_type().await.ok();
+                let mut type_val = 0;
+                if let Some(ft) = file_type {
+                    if ft.is_dir() {
+                        type_val = 4; // DT_DIR roughly
+                    } else if ft.is_file() {
+                        type_val = 8; // DT_REG roughly
+                    }
+                }
+
+                let dir_entry = DirEntry {
+                    name: entry.file_name().to_string_lossy().into_owned(),
+                    r#type: type_val,
+                };
+                
+                if tx.send(Ok(dir_entry)).await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        Ok(rx)
+    }
+
+    async fn read_file(&self, req: ReadRequest) -> Result<Receiver<Result<FileChunk, anyhow::Error>>> {
+        let full_path = self.resolve_path(&req.path)?;
+        let mut file = fs::File::open(&full_path).await.context("Failed to open file")?;
+        
+        // Seek to the requested offset
+        file.seek(std::io::SeekFrom::Start(req.offset)).await.context("Failed to seek in file")?;
+
+        // Inform the kernel about our sequential read pattern for aggressive read-ahead
+        #[cfg(target_os = "linux")]
+        {
+            use std::os::unix::io::AsRawFd;
+            let fd = file.as_raw_fd();
+            // POSIX_FADV_SEQUENTIAL = 2
+            unsafe {
+                libc::posix_fadvise(fd, req.offset as libc::off_t, req.size as libc::off_t, libc::POSIX_FADV_SEQUENTIAL);
+            }
+        }
+
+        // Limit chunks to 1MB default, or user requested, capping at 4MB for sanity
+        let mut chunk_size = req.chunk_size as usize;
+        if chunk_size == 0 {
+            chunk_size = 1024 * 1024; // 1 MB default
+        }
+        chunk_size = chunk_size.min(4 * 1024 * 1024);
+
+        // Keep the channel size small (2-4 chunks) to apply backpressure and save memory
+        let (tx, rx) = mpsc::channel(4);
+        let max_size = req.size;
+
+        tokio::spawn(async move {
+            let mut remaining = max_size;
+            use bytes::BytesMut;
+
+            while remaining > 0 {
+                let to_read = (remaining as usize).min(chunk_size);
+                let mut buffer = BytesMut::zeroed(to_read);
+
+                match file.read(&mut buffer).await {
+                    Ok(0) => break, // EOF
+                    Ok(n) => {
+                        buffer.truncate(n);
+                        let chunk = FileChunk {
+                            data: buffer.freeze().to_vec(),
+                        };
+                        remaining = remaining.saturating_sub(n as u32);
+                        if tx.send(Ok(chunk)).await.is_err() {
+                            break; // Client disconnected
+                        }
+                    }
+                    Err(e) => {
+                        let _ = tx.send(Err(e.into())).await;
+                        break;
+                    }
+                }
+            }
+        });
+
+        Ok(rx)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use tempfile::tempdir;
+
+    #[tokio::test]
+    async fn test_local_vfs_get_attr() {
+        let dir = tempdir().unwrap();
+        let file_path = dir.path().join("test.txt");
+        let mut file = std::fs::File::create(&file_path).unwrap();
+        file.write_all(b"hello world").unwrap();
+
+        let vfs = LocalVfs::new(dir.path());
+        let attr = vfs.get_attr("test.txt").await.unwrap();
+        assert_eq!(attr.size, 11);
+    }
+
+    #[tokio::test]
+    async fn test_local_vfs_read_dir() {
+        let dir = tempdir().unwrap();
+        std::fs::File::create(dir.path().join("file1.txt")).unwrap();
+        std::fs::create_dir(dir.path().join("subdir")).unwrap();
+
+        let vfs = LocalVfs::new(dir.path());
+        let mut rx = vfs.read_dir("/").await.unwrap();
+        
+        let mut entries = Vec::new();
+        while let Some(Ok(entry)) = rx.recv().await {
+            entries.push(entry.name);
+        }
+        
+        assert!(entries.contains(&"file1.txt".to_string()));
+        assert!(entries.contains(&"subdir".to_string()));
+    }
+
+    #[tokio::test]
+    async fn test_local_vfs_read_file() {
+        let dir = tempdir().unwrap();
+        let file_path = dir.path().join("data.bin");
+        let mut file = std::fs::File::create(&file_path).unwrap();
+        file.write_all(b"0123456789").unwrap(); // 10 bytes
+
+        let vfs = LocalVfs::new(dir.path());
+        let req = ReadRequest {
+            path: "data.bin".to_string(),
+            offset: 2,
+            size: 5,
+            chunk_size: 0,
+        };
+
+        let mut rx = vfs.read_file(req).await.unwrap();
+        let mut result = Vec::new();
+        while let Some(Ok(chunk)) = rx.recv().await {
+            result.extend_from_slice(&chunk.data);
+        }
+
+        assert_eq!(result, b"23456"); // bytes from offset 2, length 5
+    }
+}

furumi-server/src/vfs/mod.rs

@@ -0,0 +1,20 @@
+pub mod local;
+
+use anyhow::Result;
+use tokio::sync::mpsc::Receiver;
+
+use furumi_common::proto::{AttrResponse, DirEntry, FileChunk, ReadRequest};
+
+/// Abstract Virtual File System trait representing the remote server operations.
+#[async_trait::async_trait]
+pub trait VirtualFileSystem: Send + Sync + 'static {
+    /// Get attributes for a file or directory.
+    async fn get_attr(&self, path: &str) -> Result<AttrResponse>;
+
+    /// Read directory entries. Returns a stream (Receiver) of DirEntry.
+    /// Uses Result<DirEntry> inside the Receiver to propagate individual entry errors if any.
+    async fn read_dir(&self, path: &str) -> Result<Receiver<Result<DirEntry, anyhow::Error>>>;
+
+    /// Read chunks of a file. Returns a stream (Receiver) of FileChunk.
+    async fn read_file(&self, req: ReadRequest) -> Result<Receiver<Result<FileChunk, anyhow::Error>>>;
+}