pixelflux/webcam/
server.rs

1//! Control socket for the Selkies V4L2 interposer.
2//!
3//! Each application `open()` of the virtual device becomes one client connection. The server sends
4//! the ring configuration once, with the ring memfd attached as `SCM_RIGHTS` ancillary data, then
5//! only ever writes one-byte doorbells — one per published frame — so the interposer can block in
6//! `poll()`/`VIDIOC_DQBUF` on plain socket readability. Frame bytes never cross the socket. The
7//! interposer answers the handshake with a single byte; until that arrives a client gets no
8//! doorbells, and a closed connection is retired the moment its read end reports hangup.
9//!
10//! The accept/hangup loop runs on its own thread, woken through an eventfd for shutdown, so a
11//! process that opens the device while no frames flow is still served immediately.
12
13use std::io;
14use std::mem;
15use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
16use std::os::unix::net::UnixListener;
17use std::ptr;
18use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
19use std::sync::{Arc, Mutex};
20use std::thread::{self, JoinHandle};
21
22use super::ring::CONFIG_SIZE;
23
24struct Client {
25    fd: OwnedFd,
26    /// Set once the interposer's handshake byte arrived; only ready clients receive doorbells.
27    ready: bool,
28}
29
30struct Shared {
31    clients: Mutex<Vec<Client>>,
32    ready_count: AtomicUsize,
33    stop: AtomicBool,
34}
35
36pub struct Server {
37    path: String,
38    shared: Arc<Shared>,
39    wake: OwnedFd,
40    thread: Option<JoinHandle<()>>,
41}
42
43impl Server {
44    /// Bind `path` (replacing a stale socket file) and start serving `config` + `ring_fd`.
45    ///
46    /// A file there that another account owns is refused: in a shared directory it is that
47    /// account's listener, and every camera client of this session would reach it.
48    pub fn bind(path: &str, config: [u8; CONFIG_SIZE], ring_fd: RawFd) -> io::Result<Self> {
49        use std::os::unix::fs::MetadataExt;
50        if let Some(dir) = std::path::Path::new(path).parent()
51            && !dir.as_os_str().is_empty()
52        {
53            std::fs::create_dir_all(dir)?;
54        }
55        match std::fs::symlink_metadata(path) {
56            Ok(meta) if meta.uid() != unsafe { libc::geteuid() } => {
57                return Err(io::Error::new(
58                    io::ErrorKind::PermissionDenied,
59                    format!("{path} belongs to uid {}, not this session", meta.uid()),
60                ));
61            }
62            Ok(_) => std::fs::remove_file(path)?,
63            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
64            Err(e) => return Err(e),
65        }
66        let listener = UnixListener::bind(path)?;
67        restrict_to_owner(path)?;
68        listener.set_nonblocking(true)?;
69        let wake_raw = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
70        if wake_raw < 0 {
71            return Err(io::Error::last_os_error());
72        }
73        let wake = unsafe { OwnedFd::from_raw_fd(wake_raw) };
74        let shared = Arc::new(Shared {
75            clients: Mutex::new(Vec::new()),
76            ready_count: AtomicUsize::new(0),
77            stop: AtomicBool::new(false),
78        });
79        let t_shared = shared.clone();
80        let t_wake = wake.as_raw_fd();
81        let thread = thread::Builder::new()
82            .name("pixelflux-webcam-ctl".into())
83            .spawn(move || serve(listener, t_wake, config, ring_fd, t_shared))?;
84        Ok(Server {
85            path: path.to_string(),
86            shared,
87            wake,
88            thread: Some(thread),
89        })
90    }
91
92    pub fn path(&self) -> &str {
93        &self.path
94    }
95
96    /// Number of interposer clients that completed the handshake.
97    pub fn client_count(&self) -> usize {
98        self.shared.ready_count.load(Ordering::Relaxed)
99    }
100
101    /// Wake every ready client: one byte per published frame. A full socket buffer means a
102    /// wakeup is already pending, so `EAGAIN` is not an error; a dead peer is retired here.
103    pub fn ring_doorbell(&self) {
104        let mut clients = self
105            .shared
106            .clients
107            .lock()
108            .unwrap_or_else(|e| e.into_inner());
109        let mut i = 0;
110        while i < clients.len() {
111            if !clients[i].ready {
112                i += 1;
113                continue;
114            }
115            let byte = [1u8];
116            let n = unsafe {
117                libc::send(
118                    clients[i].fd.as_raw_fd(),
119                    byte.as_ptr() as *const _,
120                    1,
121                    libc::MSG_DONTWAIT | libc::MSG_NOSIGNAL,
122                )
123            };
124            if n < 0 {
125                let err = io::Error::last_os_error();
126                match err.raw_os_error() {
127                    Some(libc::EAGAIN) | Some(libc::EINTR) => {}
128                    _ => {
129                        clients.remove(i);
130                        self.shared.ready_count.fetch_sub(1, Ordering::Relaxed);
131                        continue;
132                    }
133                }
134            }
135            i += 1;
136        }
137    }
138
139    fn wake_thread(&self) {
140        let one: u64 = 1;
141        unsafe { libc::write(self.wake.as_raw_fd(), &one as *const u64 as *const _, 8) };
142    }
143}
144
145impl Drop for Server {
146    /// Stop accepting, close every client (the interposer sees EOF and reports `ENODEV`), and remove
147    /// the socket file.
148    fn drop(&mut self) {
149        self.shared.stop.store(true, Ordering::Release);
150        self.wake_thread();
151        if let Some(t) = self.thread.take() {
152            let _ = t.join();
153        }
154        self.shared
155            .clients
156            .lock()
157            .unwrap_or_else(|e| e.into_inner())
158            .clear();
159        self.shared.ready_count.store(0, Ordering::Relaxed);
160        let _ = std::fs::remove_file(&self.path);
161    }
162}
163
164/// Send the configuration struct with the ring memfd attached as `SCM_RIGHTS`.
165fn send_config_with_fd(sock: RawFd, config: &[u8], fd: RawFd) -> io::Result<()> {
166    let mut iov = libc::iovec {
167        iov_base: config.as_ptr() as *mut libc::c_void,
168        iov_len: config.len(),
169    };
170    let space = unsafe { libc::CMSG_SPACE(mem::size_of::<RawFd>() as u32) } as usize;
171    let mut cbuf = vec![0u64; space.div_ceil(mem::size_of::<u64>())];
172    let mut msg: libc::msghdr = unsafe { mem::zeroed() };
173    msg.msg_iov = &mut iov;
174    msg.msg_iovlen = 1;
175    msg.msg_control = cbuf.as_mut_ptr() as *mut libc::c_void;
176    msg.msg_controllen = space as _;
177    unsafe {
178        let cmsg = libc::CMSG_FIRSTHDR(&msg);
179        (*cmsg).cmsg_level = libc::SOL_SOCKET;
180        (*cmsg).cmsg_type = libc::SCM_RIGHTS;
181        (*cmsg).cmsg_len = libc::CMSG_LEN(mem::size_of::<RawFd>() as u32) as _;
182        ptr::write_unaligned(libc::CMSG_DATA(cmsg) as *mut RawFd, fd);
183    }
184    loop {
185        let n = unsafe { libc::sendmsg(sock, &msg, libc::MSG_NOSIGNAL) };
186        if n < 0 {
187            let err = io::Error::last_os_error();
188            if err.raw_os_error() == Some(libc::EINTR) {
189                continue;
190            }
191            return Err(err);
192        }
193        if n as usize != config.len() {
194            return Err(io::Error::new(
195                io::ErrorKind::WriteZero,
196                "short config send",
197            ));
198        }
199        return Ok(());
200    }
201}
202
203fn serve(
204    listener: UnixListener,
205    wake: RawFd,
206    config: [u8; CONFIG_SIZE],
207    ring_fd: RawFd,
208    shared: Arc<Shared>,
209) {
210    let listen_fd = listener.as_raw_fd();
211    let mut pollfds: Vec<libc::pollfd> = Vec::new();
212    while !shared.stop.load(Ordering::Acquire) {
213        pollfds.clear();
214        pollfds.push(libc::pollfd {
215            fd: listen_fd,
216            events: libc::POLLIN,
217            revents: 0,
218        });
219        pollfds.push(libc::pollfd {
220            fd: wake,
221            events: libc::POLLIN,
222            revents: 0,
223        });
224        {
225            let clients = shared.clients.lock().unwrap_or_else(|e| e.into_inner());
226            for c in clients.iter() {
227                pollfds.push(libc::pollfd {
228                    fd: c.fd.as_raw_fd(),
229                    events: libc::POLLIN | libc::POLLRDHUP,
230                    revents: 0,
231                });
232            }
233        }
234        let n = unsafe { libc::poll(pollfds.as_mut_ptr(), pollfds.len() as libc::nfds_t, -1) };
235        if n < 0 {
236            if io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
237                continue;
238            }
239            break;
240        }
241        if pollfds[1].revents != 0 {
242            let mut v: u64 = 0;
243            unsafe { libc::read(wake, &mut v as *mut u64 as *mut _, 8) };
244            if shared.stop.load(Ordering::Acquire) {
245                break;
246            }
247        }
248        if pollfds[0].revents != 0 {
249            accept_clients(&listener, &config, ring_fd, &shared);
250        }
251        let mut clients = shared.clients.lock().unwrap_or_else(|e| e.into_inner());
252        let mut idx = 0;
253        for pfd in pollfds.iter().skip(2) {
254            if idx >= clients.len() {
255                break;
256            }
257            if clients[idx].fd.as_raw_fd() != pfd.fd {
258                idx += 1;
259                continue;
260            }
261            if pfd.revents == 0 {
262                idx += 1;
263                continue;
264            }
265            let mut buf = [0u8; 64];
266            let r = unsafe {
267                libc::recv(
268                    clients[idx].fd.as_raw_fd(),
269                    buf.as_mut_ptr() as *mut _,
270                    buf.len(),
271                    libc::MSG_DONTWAIT,
272                )
273            };
274            let closed = if r > 0 {
275                if !clients[idx].ready {
276                    clients[idx].ready = true;
277                    shared.ready_count.fetch_add(1, Ordering::Relaxed);
278                }
279                false
280            } else if r == 0 {
281                true
282            } else {
283                !matches!(
284                    io::Error::last_os_error().raw_os_error(),
285                    Some(libc::EAGAIN) | Some(libc::EINTR)
286                )
287            };
288            if closed
289                || pfd.revents & (libc::POLLHUP | libc::POLLERR | libc::POLLRDHUP) != 0 && r <= 0
290            {
291                if clients[idx].ready {
292                    shared.ready_count.fetch_sub(1, Ordering::Relaxed);
293                }
294                clients.remove(idx);
295                continue;
296            }
297            idx += 1;
298        }
299    }
300}
301
302/// Make a socket this process bound connectable by its user alone, whatever the umask.
303pub(crate) fn restrict_to_owner(path: &str) -> io::Result<()> {
304    use std::os::unix::fs::PermissionsExt;
305    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
306}
307
308/// Whether a socket's peer runs as this process's user or as root, who passes the
309/// socket's mode anyway: one that connected before `restrict_to_owner` is refused here.
310pub(crate) fn peer_is_session_user(fd: RawFd) -> bool {
311    let mut cred: libc::ucred = unsafe { std::mem::zeroed() };
312    let mut len = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
313    let r = unsafe {
314        libc::getsockopt(
315            fd,
316            libc::SOL_SOCKET,
317            libc::SO_PEERCRED,
318            (&mut cred as *mut libc::ucred).cast(),
319            &mut len,
320        )
321    };
322    r == 0 && (cred.uid == unsafe { libc::geteuid() } || cred.uid == 0)
323}
324
325fn accept_clients(
326    listener: &UnixListener,
327    config: &[u8; CONFIG_SIZE],
328    ring_fd: RawFd,
329    shared: &Shared,
330) {
331    loop {
332        match listener.accept() {
333            Ok((stream, _)) => {
334                if !peer_is_session_user(stream.as_raw_fd()) {
335                    continue;
336                }
337                let _ = stream.set_nonblocking(true);
338                let fd = OwnedFd::from(stream);
339                match send_config_with_fd(fd.as_raw_fd(), config, ring_fd) {
340                    Ok(()) => {
341                        shared
342                            .clients
343                            .lock()
344                            .unwrap_or_else(|e| e.into_inner())
345                            .push(Client { fd, ready: false });
346                    }
347                    Err(e) => eprintln!("[webcam] interposer handshake failed: {}", e),
348                }
349            }
350            Err(e) if e.kind() == io::ErrorKind::WouldBlock => return,
351            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
352            Err(e) => {
353                eprintln!("[webcam] accept failed: {}", e);
354                return;
355            }
356        }
357    }
358}
359
360#[cfg(test)]
361mod tests {
362    use super::*;
363    use std::io::{Read, Write};
364    use std::os::unix::net::UnixStream;
365    use std::time::{Duration, Instant};
366
367    /// The socket is its user's alone whatever the umask, and a peer of that user passes.
368    #[test]
369    fn the_socket_is_connectable_by_its_user_alone() {
370        use std::os::unix::fs::PermissionsExt;
371        let dir = std::env::temp_dir().join(format!("pxwc-mode-{}", std::process::id()));
372        let _ = std::fs::remove_dir_all(&dir);
373        std::fs::create_dir(&dir).unwrap();
374        let path = dir.join("cam.sock").to_string_lossy().into_owned();
375        let old = unsafe { libc::umask(0) };
376        let listener = UnixListener::bind(&path).unwrap();
377        unsafe { libc::umask(old) };
378        restrict_to_owner(&path).unwrap();
379        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
380        assert_eq!(mode, 0o600);
381        let _client = UnixStream::connect(&path).unwrap();
382        let (peer, _) = listener.accept().unwrap();
383        assert!(peer_is_session_user(peer.as_raw_fd()));
384        std::fs::remove_dir_all(&dir).unwrap();
385    }
386
387    fn recv_config(stream: &UnixStream) -> (Vec<u8>, RawFd) {
388        let mut data = vec![0u8; CONFIG_SIZE];
389        let mut iov = libc::iovec {
390            iov_base: data.as_mut_ptr() as *mut _,
391            iov_len: data.len(),
392        };
393        let space = unsafe { libc::CMSG_SPACE(4) } as usize;
394        let mut cbuf = vec![0u64; space.div_ceil(8)];
395        let mut msg: libc::msghdr = unsafe { mem::zeroed() };
396        msg.msg_iov = &mut iov;
397        msg.msg_iovlen = 1;
398        msg.msg_control = cbuf.as_mut_ptr() as *mut _;
399        msg.msg_controllen = space as _;
400        let n = unsafe { libc::recvmsg(stream.as_raw_fd(), &mut msg, 0) };
401        assert_eq!(n as usize, CONFIG_SIZE);
402        let cmsg = unsafe { libc::CMSG_FIRSTHDR(&msg) };
403        assert!(!cmsg.is_null());
404        let fd = unsafe { ptr::read_unaligned(libc::CMSG_DATA(cmsg) as *const RawFd) };
405        (data, fd)
406    }
407
408    #[test]
409    fn handshake_fd_and_doorbells() {
410        let dir = std::env::temp_dir().join(format!("pxwc-{}", std::process::id()));
411        let path = dir.join("cam.sock");
412        let ring = crate::memfd_create(c"t", libc::MFD_CLOEXEC);
413        assert!(ring >= 0);
414        unsafe { libc::ftruncate(ring, 8192) };
415        let mut config = [0u8; CONFIG_SIZE];
416        config[..4].copy_from_slice(&0xDEADBEEFu32.to_le_bytes());
417        let server = Server::bind(path.to_str().unwrap(), config, ring).unwrap();
418
419        let mut client = UnixStream::connect(&path).unwrap();
420        let (data, fd) = recv_config(&client);
421        assert_eq!(&data[..4], &0xDEADBEEFu32.to_le_bytes());
422        assert!(fd >= 0);
423        let mut st: libc::stat = unsafe { mem::zeroed() };
424        assert_eq!(unsafe { libc::fstat(fd, &mut st) }, 0);
425        assert_eq!(st.st_size, 8192);
426        unsafe { libc::close(fd) };
427
428        server.ring_doorbell();
429        assert_eq!(server.client_count(), 0);
430        client.write_all(&[8u8]).unwrap();
431        let deadline = Instant::now() + Duration::from_secs(3);
432        while server.client_count() != 1 && Instant::now() < deadline {
433            thread::sleep(Duration::from_millis(5));
434        }
435        assert_eq!(server.client_count(), 1);
436        server.ring_doorbell();
437        client
438            .set_read_timeout(Some(Duration::from_secs(2)))
439            .unwrap();
440        let mut b = [0u8; 1];
441        client.read_exact(&mut b).unwrap();
442        assert_eq!(b[0], 1);
443
444        drop(client);
445        let deadline = Instant::now() + Duration::from_secs(3);
446        while server.client_count() != 0 && Instant::now() < deadline {
447            server.ring_doorbell();
448            thread::sleep(Duration::from_millis(5));
449        }
450        assert_eq!(server.client_count(), 0);
451        drop(server);
452        assert!(!path.exists());
453        unsafe { libc::close(ring) };
454        let _ = std::fs::remove_dir_all(&dir);
455    }
456}
pixelflux/webcam/server.rs

pixelflux/webcam/
server.rs
