wallr_core/ipc/
mod.rs

1//! Async IPC transport (Tokio Unix socket server and client).
2//! Command/response vocabulary and validation live in `wallr-common`.
3
4pub use wallr_common::ipc::*;
5
6use std::path::{Path, PathBuf};
7use std::sync::Arc;
8use std::time::Duration;
9use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
10use tokio::net::{UnixListener, UnixStream};
11use tokio::sync::Semaphore;
12
13/// Maximum response accepted from a daemon. A status/`info` response grows
14/// with the output count but is still far below this bound.
15pub const MAX_IPC_RESPONSE_BYTES: usize = 1024 * 1024;
16
17const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
18const WRITE_TIMEOUT: Duration = Duration::from_secs(5);
19const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
20const RESPONSE_TIMEOUT: Duration = Duration::from_secs(120);
21const HANDLER_TIMEOUT: Duration = Duration::from_secs(120);
22const MAX_CONCURRENT_CONNECTIONS: usize = 64;
23
24#[derive(Debug, thiserror::Error)]
25pub enum IpcError {
26    #[error("daemon not running: {0}")]
27    DaemonNotRunning(String),
28    #[error("daemon already running: {0}")]
29    AlreadyRunning(String),
30    #[error(
31        "refusing to use IPC socket path {0}: it exists but is not a socket owned by this user"
32    )]
33    UnsafeSocketPath(String),
34    #[error("IPC request/response timed out: {0}")]
35    Timeout(String),
36    #[error("IPC response exceeds {MAX_IPC_RESPONSE_BYTES} bytes")]
37    ResponseTooLarge,
38    #[error("IPC I/O error: {0}")]
39    Io(#[from] std::io::Error),
40    #[error("protocol serialization/deserialization error: {0}")]
41    Protocol(#[from] serde_json::Error),
42}
43
44/// An exclusively owned, bound Unix-domain IPC socket.
45///
46/// The advisory lock file serializes daemon startup so two processes can
47/// never both decide that the socket is stale. The guard removes the socket
48/// on drop, but only when the path still refers to the exact inode created
49/// by this instance, so a successor's socket is never unlinked.
50pub struct IpcSocket {
51    path: PathBuf,
52    listener: UnixListener,
53    dev: u64,
54    ino: u64,
55    _lock: std::fs::File,
56}
57
58impl IpcSocket {
59    /// Acquires the socket path and binds a listener.
60    ///
61    /// A live socket is reported as [`IpcError::AlreadyRunning`]. A stale
62    /// socket (connection refused) is removed and replaced. Any other kind
63    /// of filesystem object at the path is left untouched and rejected.
64    pub async fn acquire<P: AsRef<Path>>(path: P) -> Result<Self, IpcError> {
65        let path = path.as_ref().to_path_buf();
66        let lock_path = socket_lock_path(&path);
67        let lock = open_lock_file(&lock_path)?;
68        lock_exclusive(&lock).map_err(|err| {
69            if err.kind() == std::io::ErrorKind::WouldBlock {
70                IpcError::AlreadyRunning(path.to_string_lossy().to_string())
71            } else {
72                IpcError::Io(err)
73            }
74        })?;
75
76        match std::fs::symlink_metadata(&path) {
77            Ok(metadata) => {
78                use std::os::unix::fs::{FileTypeExt, MetadataExt};
79
80                if !metadata.file_type().is_socket() {
81                    return Err(IpcError::UnsafeSocketPath(
82                        path.to_string_lossy().to_string(),
83                    ));
84                }
85                // SAFETY: geteuid has no preconditions and does not touch
86                // memory; it only reads the process credentials.
87                let euid = unsafe { libc::geteuid() };
88                if metadata.uid() != euid {
89                    return Err(IpcError::UnsafeSocketPath(
90                        path.to_string_lossy().to_string(),
91                    ));
92                }
93                match UnixStream::connect(&path).await {
94                    Ok(_) => {
95                        return Err(IpcError::AlreadyRunning(path.to_string_lossy().to_string()));
96                    }
97                    Err(err)
98                        if matches!(
99                            err.kind(),
100                            std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
101                        ) =>
102                    {
103                        std::fs::remove_file(&path)?;
104                    }
105                    Err(err) => {
106                        // EACCES, EMFILE, and similar conditions are not
107                        // proof that the socket is stale, so never unlink it.
108                        return Err(IpcError::Io(err));
109                    }
110                }
111            }
112            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
113            Err(err) => return Err(IpcError::Io(err)),
114        }
115
116        let listener = UnixListener::bind(&path)?;
117        use std::os::unix::fs::{MetadataExt, PermissionsExt};
118        if let Err(err) = std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) {
119            let _ = std::fs::remove_file(&path);
120            return Err(IpcError::Io(err));
121        }
122        let metadata = match std::fs::symlink_metadata(&path) {
123            Ok(metadata) => metadata,
124            Err(err) => {
125                let _ = std::fs::remove_file(&path);
126                return Err(IpcError::Io(err));
127            }
128        };
129
130        Ok(Self {
131            path,
132            listener,
133            dev: metadata.dev(),
134            ino: metadata.ino(),
135            _lock: lock,
136        })
137    }
138
139    pub fn path(&self) -> &Path {
140        &self.path
141    }
142
143    /// Serves IPC connections until the listener fails.
144    ///
145    /// At most [`MAX_CONCURRENT_CONNECTIONS`] connections are handled at
146    /// once. Accepting is gated by the semaphore, so a flood backpressures in
147    /// the kernel backlog instead of allocating unbounded tasks.
148    pub async fn serve<F, Fut>(self, handler: F) -> Result<(), IpcError>
149    where
150        F: Fn(IpcCommand) -> Fut + Send + Sync + 'static,
151        Fut: std::future::Future<Output = IpcResponse> + Send + 'static,
152    {
153        let handler = Arc::new(handler);
154        let permits = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTIONS));
155        loop {
156            let permit = match Arc::clone(&permits).acquire_owned().await {
157                Ok(permit) => permit,
158                Err(_) => return Ok(()),
159            };
160            match self.listener.accept().await {
161                Ok((stream, _)) => {
162                    let handler = Arc::clone(&handler);
163                    tokio::spawn(async move {
164                        let _permit = permit;
165                        handle_connection(stream, handler).await;
166                    });
167                }
168                Err(err) => {
169                    tracing::error!("IPC accept error: {:?}", err);
170                    tokio::time::sleep(Duration::from_millis(100)).await;
171                }
172            }
173        }
174    }
175
176    /// Runs [`IpcSocket::serve`] on a detached task.
177    pub fn spawn<F, Fut>(self, handler: F)
178    where
179        F: Fn(IpcCommand) -> Fut + Send + Sync + 'static,
180        Fut: std::future::Future<Output = IpcResponse> + Send + 'static,
181    {
182        tokio::spawn(async move {
183            if let Err(err) = self.serve(handler).await {
184                tracing::error!("IPC server stopped: {err}");
185            }
186        });
187    }
188}
189
190/// Identity of the bound socket inode, used to clean up safely on paths
191/// where `Drop` cannot run (for example a process-wide shutdown handler).
192#[derive(Debug, Clone, Copy)]
193pub struct IpcSocketIdentity {
194    dev: u64,
195    ino: u64,
196}
197
198impl IpcSocket {
199    pub fn identity(&self) -> IpcSocketIdentity {
200        IpcSocketIdentity {
201            dev: self.dev,
202            ino: self.ino,
203        }
204    }
205}
206
207/// Removes `path` only when it is still the exact socket inode identified by
208/// `identity`. Returns whether a file was removed.
209pub fn remove_socket_if_owned(path: &Path, identity: IpcSocketIdentity) -> bool {
210    let owned = std::fs::symlink_metadata(path)
211        .is_ok_and(|metadata| existed_as_same_socket(&metadata, identity.dev, identity.ino));
212    if owned {
213        let _ = std::fs::remove_file(path);
214    }
215    owned
216}
217
218impl Drop for IpcSocket {
219    fn drop(&mut self) {
220        let _ = remove_socket_if_owned(&self.path, self.identity());
221    }
222}
223
224fn existed_as_same_socket(metadata: &std::fs::Metadata, dev: u64, ino: u64) -> bool {
225    use std::os::unix::fs::{FileTypeExt, MetadataExt};
226    metadata.file_type().is_socket() && metadata.dev() == dev && metadata.ino() == ino
227}
228
229fn socket_lock_path(path: &Path) -> PathBuf {
230    let mut name = path
231        .file_name()
232        .map(|name| name.to_os_string())
233        .unwrap_or_else(|| std::ffi::OsString::from("wallr.sock"));
234    name.push(".lock");
235    path.with_file_name(name)
236}
237
238fn open_lock_file(path: &Path) -> std::io::Result<std::fs::File> {
239    use std::os::unix::fs::OpenOptionsExt;
240    std::fs::OpenOptions::new()
241        .read(true)
242        .write(true)
243        .create(true)
244        .truncate(false)
245        .mode(0o600)
246        .open(path)
247}
248
249fn lock_exclusive(file: &std::fs::File) -> std::io::Result<()> {
250    use std::os::unix::io::AsRawFd;
251    // SAFETY: the fd belongs to `file`, which outlives this call. flock only
252    // manipulates kernel lock state for that fd and does not access memory.
253    let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
254    if result == 0 {
255        Ok(())
256    } else {
257        Err(std::io::Error::last_os_error())
258    }
259}
260
261async fn handle_connection<F, Fut>(stream: UnixStream, handler: Arc<F>)
262where
263    F: Fn(IpcCommand) -> Fut + Send + Sync + 'static,
264    Fut: std::future::Future<Output = IpcResponse> + Send + 'static,
265{
266    let (reader, mut writer) = tokio::io::split(stream);
267    let reader = BufReader::new(reader);
268    let mut limited = reader.take((MAX_IPC_BYTES + 1) as u64);
269    let mut buf = Vec::with_capacity(1024);
270    let response =
271        match tokio::time::timeout(REQUEST_TIMEOUT, limited.read_until(b'\n', &mut buf)).await {
272            Err(_) => IpcResponse::err("IPC request timed out"),
273            Ok(Ok(0)) => return,
274            Ok(Ok(_)) if buf.len() > MAX_IPC_BYTES => {
275                IpcResponse::err(format!("request exceeds {MAX_IPC_BYTES} bytes"))
276            }
277            Ok(Ok(_)) => match serde_json::from_slice::<IpcCommand>(&buf) {
278                Ok(cmd) => match validate_command(&cmd) {
279                    Ok(()) => match tokio::time::timeout(HANDLER_TIMEOUT, handler(cmd)).await {
280                        Ok(response) => response,
281                        Err(_) => IpcResponse::err("IPC command timed out"),
282                    },
283                    Err(reason) => IpcResponse::err(reason),
284                },
285                Err(e) => IpcResponse::err(format!("invalid IPC command: {e}")),
286            },
287            Ok(Err(e)) => IpcResponse::err(format!("IPC read error: {e}")),
288        };
289    write_response(&mut writer, &response).await;
290}
291
292async fn write_response(writer: &mut tokio::io::WriteHalf<UnixStream>, response: &IpcResponse) {
293    let Ok(res_data) = serde_json::to_vec(response) else {
294        return;
295    };
296    let write = async {
297        writer.write_all(&res_data).await?;
298        writer.write_all(b"\n").await?;
299        writer.flush().await
300    };
301    let _ = tokio::time::timeout(WRITE_TIMEOUT, write).await;
302}
303
304async fn read_response(reader: &mut BufReader<UnixStream>) -> Result<Vec<u8>, IpcError> {
305    let mut limited = reader.take((MAX_IPC_RESPONSE_BYTES + 1) as u64);
306    let mut response_line = Vec::new();
307    let read = limited.read_until(b'\n', &mut response_line);
308    match tokio::time::timeout(RESPONSE_TIMEOUT, read).await {
309        Ok(Ok(0)) => Err(IpcError::Protocol(serde_json::Error::io(
310            std::io::Error::new(
311                std::io::ErrorKind::UnexpectedEof,
312                "daemon closed the connection without a response",
313            ),
314        ))),
315        Ok(Ok(_)) if response_line.len() > MAX_IPC_RESPONSE_BYTES => {
316            Err(IpcError::ResponseTooLarge)
317        }
318        Ok(Ok(_)) => Ok(response_line),
319        Ok(Err(err)) => Err(IpcError::Io(err)),
320        Err(_) => Err(IpcError::Timeout("waiting for daemon response".to_string())),
321    }
322}
323
324pub async fn send_ipc_command<P: AsRef<Path>>(
325    socket_path: P,
326    command: IpcCommand,
327) -> Result<IpcResponse, IpcError> {
328    let socket_path = socket_path.as_ref();
329    let mut stream =
330        match tokio::time::timeout(CONNECT_TIMEOUT, UnixStream::connect(socket_path)).await {
331            Ok(Ok(stream)) => stream,
332            Ok(Err(err)) => return Err(IpcError::DaemonNotRunning(err.to_string())),
333            Err(_) => {
334                return Err(IpcError::Timeout(format!(
335                    "connecting to {}",
336                    socket_path.display()
337                )));
338            }
339        };
340
341    let req_data = serde_json::to_vec(&command)?;
342    let write = async {
343        stream.write_all(&req_data).await?;
344        stream.write_all(b"\n").await?;
345        stream.flush().await
346    };
347    match tokio::time::timeout(WRITE_TIMEOUT, write).await {
348        Ok(Ok(())) => {}
349        Ok(Err(err)) => return Err(IpcError::Io(err)),
350        Err(_) => return Err(IpcError::Timeout("sending IPC request".to_string())),
351    }
352
353    let mut reader = BufReader::new(stream);
354    let response_line = read_response(&mut reader).await?;
355    let response: IpcResponse = serde_json::from_slice(&response_line)?;
356    Ok(response)
357}
358
359/// Bundled server helper kept for callers that do not need to reserve the
360/// socket before setting up the rest of the daemon.
361pub async fn start_ipc_server<P, F, Fut>(socket_path: P, handler: F) -> Result<(), IpcError>
362where
363    P: AsRef<Path>,
364    F: Fn(IpcCommand) -> Fut + Send + Sync + 'static,
365    Fut: std::future::Future<Output = IpcResponse> + Send + 'static,
366{
367    IpcSocket::acquire(socket_path).await?.spawn(handler);
368    Ok(())
369}
370
371#[cfg(test)]
372mod tests {
373    use super::*;
374    use std::os::unix::fs::PermissionsExt;
375
376    #[tokio::test]
377    async fn socket_is_private_and_removed_on_drop() {
378        let temp = tempfile::tempdir().unwrap();
379        let path = temp.path().join("wallr.sock");
380        let socket = IpcSocket::acquire(&path).await.unwrap();
381        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
382        assert_eq!(mode, 0o600);
383        drop(socket);
384        assert!(!path.exists());
385    }
386
387    #[tokio::test]
388    async fn live_socket_blocks_a_second_daemon() {
389        let temp = tempfile::tempdir().unwrap();
390        let path = temp.path().join("wallr.sock");
391        let _first = IpcSocket::acquire(&path).await.unwrap();
392        assert!(matches!(
393            IpcSocket::acquire(&path).await,
394            Err(IpcError::AlreadyRunning(_))
395        ));
396    }
397
398    #[tokio::test]
399    async fn stale_socket_is_replaced_but_regular_files_are_preserved() {
400        let temp = tempfile::tempdir().unwrap();
401        let path = temp.path().join("wallr.sock");
402        drop(std::os::unix::net::UnixListener::bind(&path).unwrap());
403        assert!(path.exists());
404        let socket = IpcSocket::acquire(&path).await.unwrap();
405        drop(socket);
406        assert!(!path.exists());
407
408        let regular = temp.path().join("not-a-socket");
409        std::fs::write(&regular, b"user data").unwrap();
410        assert!(matches!(
411            IpcSocket::acquire(&regular).await,
412            Err(IpcError::UnsafeSocketPath(_))
413        ));
414        assert_eq!(std::fs::read(&regular).unwrap(), b"user data");
415    }
416
417    #[tokio::test]
418    async fn server_round_trips_commands_and_rejects_empty_preview_paths() {
419        let temp = tempfile::tempdir().unwrap();
420        let path = temp.path().join("wallr.sock");
421        let socket = IpcSocket::acquire(&path).await.unwrap();
422        let server_path = path.clone();
423        socket.spawn(|cmd| async move {
424            match cmd {
425                IpcCommand::Status => IpcResponse {
426                    success: true,
427                    message: Some("ok".into()),
428                },
429                other => IpcResponse::err(format!("unexpected {other:?}")),
430            }
431        });
432        let response = send_ipc_command(&server_path, IpcCommand::Status)
433            .await
434            .unwrap();
435        assert!(response.success);
436        assert_eq!(response.message.as_deref(), Some("ok"));
437
438        let response = send_ipc_command(
439            &server_path,
440            IpcCommand::Preview {
441                path: String::new(),
442                effect: None,
443                duration_ms: None,
444                fps: None,
445                no_theme: true,
446                theme_override: None,
447                monitor: None,
448                scaling_mode: None,
449            },
450        )
451        .await
452        .unwrap();
453        assert!(!response.success);
454    }
455}
wallr_core/ipc/mod.rs

wallr_core/ipc/
mod.rs
