Add FileTransfer (drag & drop / copy & paste files)

Host->Guest is easy; Guest->Host is not there yet because we have to do
something for the (O_PATH) fd passing such as converting to inode numbers
and getting an O_PATH fd back through the VMM, which requires modifying
libkrun/muvm to have a connection channel.. Or we should switch this from
vsock to a virtgpu channel where it would be easier to handle.

sidebus-broker/src/main.rs

@@ -144,6 +144,12 @@ async fn main() -> eyre::Result<()> {
     let priv_bus_conn = priv_bus.lock().await.connect_channel(false).await?;
     let host_session_conn = zbus::connection::Builder::session()?.build().await?;
     let file_chooser_imp = portal::file_chooser::FileChooser::new(
+        &host_session_conn,
+        &priv_bus_conn,
+        cli.guest_mountpoint.clone(),
+    )
+    .await?;
+    let file_transfer_imp = portal::file_transfer::FileTransfer::new(
         &host_session_conn,
         &priv_bus_conn,
         cli.guest_mountpoint,
@@ -154,6 +160,7 @@ async fn main() -> eyre::Result<()> {
     async fn on_vm_bus_connected(
         vm_bus_conn: zbus::Connection,
         file_chooser: portal::file_chooser::FileChooser,
+        file_transfer: portal::file_transfer::FileTransfer,
         settings: portal::settings::Settings,
     ) -> Result<(), eyre::Report> {
         if !vm_bus_conn
@@ -163,6 +170,28 @@ async fn main() -> eyre::Result<()> {
         {
             error!("org.freedesktop.portal.FileChooser already provided");
         };
+
+        if !vm_bus_conn
+            .object_server()
+            .at("/org/freedesktop/portal/documents", file_transfer)
+            .await?
+        {
+            error!("org.freedesktop.portal.FileTransfer already provided");
+        };
+        let file_transfer_ref = vm_bus_conn
+            .object_server()
+            .interface::<_, portal::file_transfer::FileTransfer>(
+                "/org/freedesktop/portal/documents",
+            )
+            .await?;
+        tokio::spawn(async move {
+            let file_transfer = file_transfer_ref.get().await;
+            let emitter = file_transfer_ref.signal_emitter();
+            if let Err(err) = file_transfer.forward_transfer_closed(emitter.clone()).await {
+                error!(%err, "forwarding forward_transfer_closed changes ended");
+            }
+        });
+
         if !vm_bus_conn
             .object_server()
             .at("/org/freedesktop/portal/desktop", settings)
@@ -181,21 +210,23 @@ async fn main() -> eyre::Result<()> {
                 error!(%err, "forwarding settings changes ended");
             }
         });
+
         // XXX: no method for "wait until the conn dies"?
         Ok(std::future::pending::<()>().await)
     }
 
     if let Some(path) = cli.unix_path {
         let vm_unix_listener = UnixListener::bind(path)?;
-        server_tasks.spawn(enclose!((file_chooser_imp, settings_imp) async move {
+        server_tasks.spawn(enclose!((file_chooser_imp, file_transfer_imp, settings_imp) async move {
             while let Ok((socket, remote_addr)) = vm_unix_listener.accept().await {
-                let f = enclose!((file_chooser_imp, settings_imp) async move {
+                let f = enclose!((file_chooser_imp, file_transfer_imp, settings_imp) async move {
                     let client_conn = zbus::connection::Builder::unix_stream(socket)
                         .auth_mechanism(zbus::AuthMechanism::Anonymous)
                         .name("org.freedesktop.portal.Desktop")?
+                        .name("org.freedesktop.portal.Documents")?
                         .build()
                         .await?;
-                    on_vm_bus_connected(client_conn, file_chooser_imp, settings_imp).await
+                    on_vm_bus_connected(client_conn, file_chooser_imp, file_transfer_imp, settings_imp).await
                 });
                 tokio::spawn(
                     async {
@@ -216,10 +247,10 @@ async fn main() -> eyre::Result<()> {
             vsock::ListenerBuilder::new(vsock::VsockAddr::new(vsock::VMADDR_CID_HOST, port))
                 .with_label("VM Bus")
                 .listen(move |client| {
-                    enclose!((file_chooser_imp, settings_imp) async move {
+                    enclose!((file_chooser_imp, file_transfer_imp, settings_imp) async move {
                         // TODO: Not necessary to go through the channel, add vsock support to the Peer too?
                         let client_conn = client.build().await?;
-                        on_vm_bus_connected(client_conn, file_chooser_imp, settings_imp).await
+                        on_vm_bus_connected(client_conn, file_chooser_imp, file_transfer_imp, settings_imp).await
                     })
                 })
                 .map_ok_or_else(

sidebus-broker/src/portal/file_chooser.rs

@@ -53,6 +53,7 @@ impl FileChooser {
                 docs: self.docs.clone(),
                 guest_root: self.guest_root.clone(),
                 for_save: false,
+                persistent: true,
                 directory: options.get_as("directory")?.unwrap_or(false),
             })
             .perform(async || self.host.open_file(parent_window, title, options).await)
@@ -73,6 +74,7 @@ impl FileChooser {
                 docs: self.docs.clone(),
                 guest_root: self.guest_root.clone(),
                 for_save: true,
+                persistent: true,
                 directory: false,
             })
             .perform(async || self.host.save_file(parent_window, title, options).await)
@@ -93,6 +95,7 @@ impl FileChooser {
                 docs: self.docs.clone(),
                 guest_root: self.guest_root.clone(),
                 for_save: true,
+                persistent: true,
                 directory: false,
             })
             .perform(async || self.host.save_files(parent_window, title, options).await)
@@ -106,11 +109,13 @@ impl FileChooser {
     }
 }
 
-struct FileTransformer {
+#[derive(Clone)]
-    docs: DocumentsProxy<'static>,
+pub struct FileTransformer {
-    guest_root: PathBuf,
+    pub docs: DocumentsProxy<'static>,
-    for_save: bool,
+    pub guest_root: PathBuf,
-    directory: bool,
+    pub for_save: bool,
+    pub persistent: bool,
+    pub directory: bool,
 }
 
 // ref: send_response_in_thread_func
@@ -134,6 +139,14 @@ impl ResultTransformer for FileTransformer {
             .async_map(|u| self.add_path_as_doc(u))
             .await
             .flatten()
+            .map(|path| match url::Url::from_file_path(&path) {
+                Ok(url) => Some(url.to_string()),
+                Err(err) => {
+                    warn!(?err, ?path, "could not make url from returned path");
+                    None
+                }
+            })
+            .flatten()
             .collect::<Vec<_>>();
 
         results.insert("uris", guest_uris.into());
@@ -150,7 +163,7 @@ const DIRECTORY: u32 = 1 << 3;
 // https://github.com/flatpak/xdg-desktop-portal/blob/10e712e06aa8eb9cd0e59c73c5be62ba53e981a4/src/xdp-documents.c#L71
 
 impl FileTransformer {
-    async fn add_path_as_doc(&self, path: PathBuf) -> Option<String> {
+    pub async fn add_path_as_doc(&self, path: PathBuf) -> Option<PathBuf> {
         use rustix::fs::{Mode, OFlags};
 
         let o_path_fd = match rustix::fs::open(
@@ -166,8 +179,8 @@ impl FileTransformer {
         };
 
         let flags = REUSE_EXISTING
-            | PERSISTENT
             | AS_NEEDED_BY_APP
+            | if self.persistent { PERSISTENT } else { 0 }
             | if self.directory { DIRECTORY } else { 0 };
 
         // XXX: portal impl can return writable=false but host frontend does not pass that back..
@@ -212,14 +225,7 @@ impl FileTransformer {
                 return None;
             }
         };
-        let path = self.guest_root.join(doc_id).join(filename);
+        Some(self.guest_root.join(doc_id).join(filename))
-        match url::Url::from_file_path(&path) {
-            Ok(url) => Some(url.to_string()),
-            Err(err) => {
-                warn!(?err, ?path, "could not make url from returned path");
-                None
-            }
-        }
     }
 }
 

sidebus-broker/src/portal/file_transfer.rs

@@ -0,0 +1,157 @@
+use std::{collections::HashMap, path::PathBuf};
+
+use tokio::sync::broadcast;
+use tokio_stream::StreamExt;
+use tracing::{debug, error};
+use zbus::{
+    Connection, fdo::Result, names::OwnedUniqueName, object_server::SignalEmitter, zvariant,
+};
+
+use super::{documents::DocumentsProxy, file_chooser::FileTransformer};
+
+#[derive(Clone)]
+pub struct FileTransfer {
+    host: FileTransferProxy<'static>,
+    file_transformer: FileTransformer,
+    tx: broadcast::Sender<ForwarderCommand>,
+}
+
+#[derive(Clone, Debug)]
+enum ForwarderCommand {
+    Add(String, OwnedUniqueName),
+    // Remove(String),
+}
+
+#[zbus::interface(
+    name = "org.freedesktop.portal.FileTransfer",
+    proxy(
+        default_service = "org.freedesktop.portal.Documents",
+        default_path = "/org/freedesktop/portal/documents"
+    )
+)]
+impl FileTransfer {
+    async fn add_files(
+        &self,
+        _key: &str,
+        _fds: Vec<zbus::zvariant::Fd<'_>>,
+        _options: HashMap<&str, zvariant::Value<'_>>,
+    ) -> Result<()> {
+        Err(zbus::fdo::Error::NotSupported(
+            "Adding files to transfer out is not yet implemented".to_owned(),
+        ))
+    }
+
+    async fn retrieve_files(
+        &self,
+        key: &str,
+        options: HashMap<&str, zvariant::Value<'_>>,
+    ) -> Result<Vec<String>> {
+        let host_paths = self.host.retrieve_files(key, options).await?;
+        let mut result = Vec::with_capacity(host_paths.len());
+        for host_path in host_paths {
+            if let Some(guest_path) = self
+                .file_transformer
+                .add_path_as_doc(PathBuf::from(&host_path))
+                .await
+            {
+                result.push(guest_path.to_string_lossy().into_owned());
+            } else {
+                debug!("could not expose path {host_path}");
+            }
+        }
+        Ok(result)
+    }
+
+    async fn start_transfer(
+        &self,
+        #[zbus(header)] hdr: zbus::message::Header<'_>,
+        options: HashMap<&str, zvariant::Value<'_>>,
+    ) -> Result<String> {
+        let sender = hdr
+            .sender()
+            .ok_or_else(|| zbus::Error::MissingField)?
+            .to_owned();
+        let key = self.host.start_transfer(options).await?;
+        debug!("start_transfer: {key}");
+        if let Err(err) = self
+            .tx
+            .send(ForwarderCommand::Add(key.clone(), sender.into()))
+        {
+            error!("file_transfer internal channel error: {err:?}");
+            return Err(zbus::fdo::Error::IOError("channel error".to_owned()));
+        }
+        Ok(key)
+    }
+
+    async fn stop_transfer(&self, key: &str) -> Result<()> {
+        debug!("stop_transfer: {key}");
+        self.host.stop_transfer(key).await
+    }
+
+    #[zbus(signal)]
+    async fn transfer_closed(signal_emitter: &SignalEmitter<'_>, key: &str) -> zbus::Result<()>;
+
+    #[zbus(property, name = "version")]
+    fn version(&self) -> Result<u32> {
+        Ok(1)
+    }
+}
+
+impl FileTransfer {
+    pub async fn new(
+        host_session_conn: &Connection,
+        priv_conn: &Connection,
+        guest_root: PathBuf,
+    ) -> Result<Self> {
+        let host = FileTransferProxy::builder(host_session_conn)
+            .build()
+            .await?;
+        let docs = DocumentsProxy::builder(priv_conn).build().await?;
+        let file_transformer = FileTransformer {
+            docs,
+            guest_root,
+            for_save: false,
+            persistent: false,
+            directory: false,
+        };
+        let (tx, _) = broadcast::channel(8);
+        Ok(FileTransfer {
+            host,
+            file_transformer,
+            tx,
+        })
+    }
+
+    pub async fn forward_transfer_closed(
+        &self,
+        mut signal_emitter: SignalEmitter<'static>,
+    ) -> Result<()> {
+        let mut stream = self.host.receive_transfer_closed().await?;
+        let mut cmds = self.tx.subscribe();
+        let mut receivers = HashMap::new();
+
+        loop {
+            tokio::select! {
+                Ok(cmd) = cmds.recv() => match cmd {
+                    ForwarderCommand::Add(key, receiver) => { receivers.insert(key, receiver); },
+                    // ForwarderCommand::Remove(key) => { receivers.remove(&key); },
+                },
+                Some(signal) = stream.next() => {
+                    debug!("transfer closed {signal:?}");
+                    if let Ok((key,)) = signal.0.deserialize::<(&str,)>() {
+                        if let Some(bus_name) = receivers.remove(key) {
+                            signal_emitter = signal_emitter.set_destination(zbus::names::BusName::Unique(bus_name.clone().into()));
+                            if let Err(err) = FileTransfer::transfer_closed(&signal_emitter, key).await {
+                                error!("could not forward signal for key {key}: {err:?}");
+                            }
+                        } else {
+                            error!("got a signal for unknown key {key}");
+                        }
+                    } else {
+                        error!("could not deserialize transfer closed signal");
+                    };
+                }
+            }
+        }
+    }
+}

sidebus-broker/src/portal/mod.rs

@@ -1,4 +1,5 @@
 pub mod documents;
 pub mod file_chooser;
+pub mod file_transfer;
 pub mod request;
 pub mod settings;

sidebus-broker/src/vsock.rs

@@ -17,6 +17,7 @@ impl ConnectionBuilder {
         zbus::connection::Builder::vsock_stream(self.socket)
             .auth_mechanism(zbus::AuthMechanism::Anonymous)
             .name("org.freedesktop.portal.Desktop")?
+            .name("org.freedesktop.portal.Documents")?
             .build()
             .await
             .map_err(|e| e.into())