muvm-guest: pwbridge: forward fds on send

Any outgoing Pipewire message with num_fd > 0 previously hit
unimplemented!(), which in practice crashed the guest on the first
ClientNode.PortBuffers that any pw-stream / pw-filter client emits.

Handle fd-bearing send messages by draining the expected fds from
request_fds, converting each via vgpu_id_from_prime into a
CrossDomainResource + GemHandleFinalizer, and attaching them to the
StreamSendResult. If request_fds is short of what the header claims,
warn and return EIO instead of panicking.

Only ClientNode.PortBuffers is treated as fd-bearing on the send path.
Per the PipeWire 1.6 native protocol that is the only client->server
method whose fds are dma-buf candidates — each fd corresponds to an
SPA_Data in the submitted buffer pool. SecurityContext.Create is the
only other send-side method that carries fds (a unix socket + an
eventfd, neither of which are dma-bufs); it is not supported by this
bridge. Any other fd-bearing send message — a future protocol
addition, an unsupported interface, or a malformed client — warns and
returns EIO rather than blindly handing the fd to
DRM_IOCTL_PRIME_FD_TO_HANDLE, which would EINVAL on anything that
isn't a dma-buf and yield an opaque failure.

Replace the unreachable!() PipeWireResourceFinalizer stub with an
enum that dispatches to GemHandleFinalizer, so the finalizers
produced here actually run.

Signed-off-by: Adam Ford <[email protected]>

Author: Adam Ford

crates/muvm/src/guest/bridge/pipewire.rs

@@ -4,15 +4,15 @@ use std::os::fd::{AsFd, AsRawFd, OwnedFd};
 use std::{env, mem};
 
 use anyhow::Result;
-use log::debug;
+use log::{debug, warn};
 use nix::errno::Errno;
 use nix::sys::epoll::EpollFlags;
 use nix::sys::eventfd::{EfdFlags, EventFd};
 
 use crate::guest::bridge::common;
 use crate::guest::bridge::common::{
-    Client, CrossDomainHeader, CrossDomainResource, MessageResourceFinalizer, ProtocolHandler,
-    StreamRecvResult, StreamSendResult,
+    Client, CrossDomainHeader, CrossDomainResource, GemHandleFinalizer, MessageResourceFinalizer,
+    ProtocolHandler, StreamRecvResult, StreamSendResult,
 };
 
 const CROSS_DOMAIN_CHANNEL_TYPE_PW: u32 = 0x10;
@@ -27,6 +27,7 @@ const PW_OPC_CORE_CREATE_OBJECT: u8 = 6;
 const PW_OPC_CORE_ADD_MEM: u8 = 6;
 const PW_OPC_CLIENT_UPDATE_PROPERTIES: u8 = 2;
 const PW_OPC_CLIENT_NODE_TRANSPORT: u8 = 0;
+const PW_OPC_CLIENT_NODE_PORT_BUFFERS: u8 = 5;
 const PW_OPC_CLIENT_NODE_SET_ACTIVATION: u8 = 10;
 
 #[repr(C)]
@@ -165,13 +166,17 @@ impl PipeWireHeader {
     }
 }
 
-struct PipeWireResourceFinalizer;
+enum PipeWireResourceFinalizer {
+    Gem(GemHandleFinalizer),
+}
 
 impl MessageResourceFinalizer for PipeWireResourceFinalizer {
     type Handler = PipeWireProtocolHandler;
 
-    fn finalize(self, _: &mut Client<Self::Handler>) -> Result<()> {
-        unreachable!()
+    fn finalize(self, client: &mut Client<Self::Handler>) -> Result<()> {
+        match self {
+            Self::Gem(fin) => fin.finalize(client),
+        }
     }
 }
 
@@ -349,13 +354,41 @@ impl ProtocolHandler for PipeWireProtocolHandler {
                     .insert(msg.new_id, ClientNodeData::new());
             }
         }
+        let mut resources = Vec::with_capacity(hdr.num_fd);
+        let mut finalizers = Vec::with_capacity(hdr.num_fd);
         if hdr.num_fd != 0 {
-            unimplemented!();
+            let is_port_buffers = hdr.opcode == PW_OPC_CLIENT_NODE_PORT_BUFFERS
+                && this.protocol_handler.client_nodes.contains_key(&hdr.id);
+            if !is_port_buffers {
+                warn!(
+                    id = hdr.id,
+                    opcode = hdr.opcode,
+                    num_fd = hdr.num_fd;
+                    "Pipewire send: unexpected fd-bearing message",
+                );
+                return Err(Errno::EIO.into());
+            }
+            if this.request_fds.len() < hdr.num_fd {
+                warn!(
+                    id = hdr.id,
+                    opcode = hdr.opcode,
+                    expected_fds = hdr.num_fd,
+                    queued_fds = this.request_fds.len();
+                    "Pipewire send: short fd queue",
+                );
+                return Err(Errno::EIO.into());
+            }
+            let fds: Vec<OwnedFd> = this.request_fds.drain(..hdr.num_fd).collect();
+            for fd in fds {
+                let (res, finalizer) = this.vgpu_id_from_prime(fd)?;
+                resources.push(res);
+                finalizers.push(PipeWireResourceFinalizer::Gem(finalizer));
+            }
         };
         Ok(StreamSendResult::Processed {
             consumed_bytes: hdr.size,
-            resources: Vec::new(),
-            finalizers: Vec::new(),
+            resources,
+            finalizers,
         })
     }