blit_server/ipc/
ipc_unix.rs

1use std::os::unix::fs::PermissionsExt;
2use std::os::unix::io::RawFd;
3use tokio::io::unix::AsyncFd;
4use tokio::net::UnixListener;
5
6pub type IpcStream = tokio::net::UnixStream;
7
8pub fn default_ipc_path() -> String {
9    if let Ok(dir) = std::env::var("TMPDIR") {
10        return format!("{dir}/blit.sock");
11    }
12    if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") {
13        return format!("{dir}/blit.sock");
14    }
15    if let Ok(user) = std::env::var("USER") {
16        return format!("/tmp/blit-{user}.sock");
17    }
18    "/tmp/blit.sock".into()
19}
20
21pub struct IpcListener {
22    inner: UnixListener,
23    /// Held for the process lifetime so the flock is released on exit.
24    _lock: Option<std::fs::File>,
25}
26
27impl IpcListener {
28    pub fn bind(path: &str, verbose: bool) -> Self {
29        // Acquire an exclusive flock on a companion lockfile so that a
30        // previous server instance is terminated before we bind.  The OS
31        // releases the lock automatically when the holder exits — even on
32        // SIGKILL — so there are no stale-lock issues.
33        let lock_path = format!("{path}.lock");
34        #[allow(clippy::suspicious_open_options)]
35        let lock_file = std::fs::OpenOptions::new()
36            .create(true)
37            .read(true)
38            .write(true)
39            .open(&lock_path)
40            .unwrap_or_else(|e| {
41                eprintln!("blit-server: cannot open {lock_path}: {e}");
42                std::process::exit(1);
43            });
44
45        use std::os::unix::io::AsRawFd;
46        let fd = lock_file.as_raw_fd();
47        let mut locked = unsafe { libc::flock(fd, libc::LOCK_EX | libc::LOCK_NB) } == 0;
48
49        if !locked {
50            // Lock held by another server — read its PID and terminate it.
51            let mut pid_str = String::new();
52            if std::io::Read::read_to_string(&mut (&lock_file), &mut pid_str).is_ok()
53                && let Ok(old_pid) = pid_str.trim().parse::<i32>()
54            {
55                eprintln!("blit-server: terminating previous server (pid {old_pid})");
56                unsafe { libc::kill(old_pid, libc::SIGTERM) };
57            }
58            // Wait up to 3 s for the old server to release the lock.
59            for _ in 0..30 {
60                std::thread::sleep(std::time::Duration::from_millis(100));
61                if unsafe { libc::flock(fd, libc::LOCK_EX | libc::LOCK_NB) } == 0 {
62                    locked = true;
63                    break;
64                }
65            }
66            if !locked {
67                eprintln!(
68                    "blit-server: cannot acquire lock {lock_path} — is another server running?"
69                );
70                std::process::exit(1);
71            }
72        }
73
74        // Record our PID so the next server instance can terminate us.
75        {
76            use std::io::{Seek, Write};
77            let _ = lock_file.set_len(0);
78            let _ = (&lock_file).seek(std::io::SeekFrom::Start(0));
79            let _ = write!(&lock_file, "{}", std::process::id());
80        }
81
82        let _ = std::fs::remove_file(path);
83        // Set a restrictive umask before bind so the socket is created with
84        // 0700 permissions atomically, closing the race window between bind
85        // and the subsequent chmod.
86        let old_umask = unsafe { libc::umask(0o077) };
87        let listener = UnixListener::bind(path).unwrap_or_else(|e| {
88            unsafe { libc::umask(old_umask) };
89            eprintln!("blit-server: cannot bind to {path}: {e}");
90            std::process::exit(1);
91        });
92        unsafe { libc::umask(old_umask) };
93        if let Err(e) = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)) {
94            eprintln!("blit-server: warning: cannot set socket permissions: {e}");
95        }
96        if verbose {
97            eprintln!("listening on {path}");
98        }
99        Self {
100            inner: listener,
101            _lock: Some(lock_file),
102        }
103    }
104
105    pub fn from_systemd_fd(verbose: bool) -> Option<Self> {
106        let fds = std::env::var("LISTEN_FDS").ok()?;
107        if fds.trim() != "1" {
108            if verbose {
109                eprintln!("LISTEN_FDS={fds}, expected 1; falling back to bind");
110            }
111            return None;
112        }
113        let pid = std::env::var("LISTEN_PID").ok()?;
114        if pid.trim() != std::process::id().to_string() {
115            if verbose {
116                eprintln!(
117                    "LISTEN_PID={pid} does not match our pid {}; falling back to bind",
118                    std::process::id()
119                );
120            }
121            return None;
122        }
123        use std::os::unix::io::FromRawFd;
124        let std_listener = unsafe { std::os::unix::net::UnixListener::from_raw_fd(3) };
125        std_listener.set_nonblocking(true).unwrap();
126        if verbose {
127            eprintln!("using socket activation (fd 3)");
128        }
129        Some(Self {
130            inner: UnixListener::from_std(std_listener).unwrap(),
131            _lock: None,
132        })
133    }
134
135    pub async fn accept(&self) -> std::io::Result<IpcStream> {
136        let (stream, _) = self.inner.accept().await?;
137        Ok(stream)
138    }
139}
140
141enum RecvFdResult {
142    Fd(RawFd),
143    WouldBlock,
144    Closed,
145}
146
147fn recv_fd(channel: RawFd) -> RecvFdResult {
148    unsafe {
149        let mut buf = [0u8; 1];
150        let mut iov = libc::iovec {
151            iov_base: buf.as_mut_ptr() as *mut libc::c_void,
152            iov_len: buf.len(),
153        };
154        let cmsg_space = libc::CMSG_SPACE(std::mem::size_of::<RawFd>() as u32) as usize;
155        let mut cmsg_buf = vec![0u8; cmsg_space];
156        let mut msg: libc::msghdr = std::mem::zeroed();
157        msg.msg_iov = &mut iov;
158        msg.msg_iovlen = 1;
159        msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void;
160        msg.msg_controllen = cmsg_space as _;
161        let n = libc::recvmsg(channel, &mut msg, libc::MSG_DONTWAIT);
162        if n < 0 {
163            let err = std::io::Error::last_os_error();
164            if err.kind() == std::io::ErrorKind::WouldBlock {
165                return RecvFdResult::WouldBlock;
166            }
167            if err.raw_os_error() == Some(libc::EINTR) {
168                return RecvFdResult::WouldBlock;
169            }
170            return RecvFdResult::Closed;
171        }
172        if n == 0 {
173            return RecvFdResult::Closed;
174        }
175        let cmsg = libc::CMSG_FIRSTHDR(&msg);
176        if cmsg.is_null() {
177            return RecvFdResult::Closed;
178        }
179        if (*cmsg).cmsg_level == libc::SOL_SOCKET && (*cmsg).cmsg_type == libc::SCM_RIGHTS {
180            let fd_ptr = libc::CMSG_DATA(cmsg) as *const RawFd;
181            RecvFdResult::Fd(std::ptr::read_unaligned(fd_ptr))
182        } else {
183            RecvFdResult::Closed
184        }
185    }
186}
187
188pub async fn run_fd_channel(channel_fd: RawFd, state: crate::AppState) {
189    use std::os::unix::io::FromRawFd;
190    if state.config.verbose {
191        eprintln!("accepting clients via fd-channel (fd {channel_fd})");
192    }
193    let channel = unsafe { std::os::unix::net::UnixStream::from_raw_fd(channel_fd) };
194    channel.set_nonblocking(true).unwrap();
195    let async_channel = AsyncFd::new(channel).unwrap();
196    loop {
197        let mut guard = match async_channel.readable().await {
198            Ok(g) => g,
199            Err(e) => {
200                eprintln!("fd-channel error: {e}");
201                break;
202            }
203        };
204        match recv_fd(channel_fd) {
205            RecvFdResult::Fd(client_fd) => {
206                let std_stream = unsafe { std::os::unix::net::UnixStream::from_raw_fd(client_fd) };
207                std_stream.set_nonblocking(true).unwrap();
208                let stream = tokio::net::UnixStream::from_std(std_stream).unwrap();
209                let state = state.clone();
210                tokio::spawn(crate::handle_client(stream, state));
211                guard.retain_ready();
212            }
213            RecvFdResult::WouldBlock => {
214                guard.clear_ready();
215            }
216            RecvFdResult::Closed => {
217                break;
218            }
219        }
220    }
221    if state.config.verbose {
222        eprintln!("fd-channel closed, shutting down");
223    }
224}
blit_server/ipc/ipc_unix.rs

blit_server/ipc/
ipc_unix.rs
