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}
24
25impl IpcListener {
26    pub fn bind(path: &str, verbose: bool) -> Self {
27        let _ = std::fs::remove_file(path);
28        let listener = UnixListener::bind(path).unwrap_or_else(|e| {
29            eprintln!("blit-server: cannot bind to {path}: {e}");
30            std::process::exit(1);
31        });
32        if let Err(e) = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700)) {
33            eprintln!("blit-server: warning: cannot set socket permissions: {e}");
34        }
35        if verbose {
36            eprintln!("listening on {path}");
37        }
38        Self { inner: listener }
39    }
40
41    pub fn from_systemd_fd(verbose: bool) -> Option<Self> {
42        let fds = std::env::var("LISTEN_FDS").ok()?;
43        if fds.trim() != "1" {
44            if verbose {
45                eprintln!("LISTEN_FDS={fds}, expected 1; falling back to bind");
46            }
47            return None;
48        }
49        use std::os::unix::io::FromRawFd;
50        let std_listener = unsafe { std::os::unix::net::UnixListener::from_raw_fd(3) };
51        std_listener.set_nonblocking(true).unwrap();
52        if verbose {
53            eprintln!("using socket activation (fd 3)");
54        }
55        Some(Self {
56            inner: UnixListener::from_std(std_listener).unwrap(),
57        })
58    }
59
60    pub async fn accept(&self) -> std::io::Result<IpcStream> {
61        let (stream, _) = self.inner.accept().await?;
62        Ok(stream)
63    }
64}
65
66enum RecvFdResult {
67    Fd(RawFd),
68    WouldBlock,
69    Closed,
70}
71
72fn recv_fd(channel: RawFd) -> RecvFdResult {
73    unsafe {
74        let mut buf = [0u8; 1];
75        let mut iov = libc::iovec {
76            iov_base: buf.as_mut_ptr() as *mut libc::c_void,
77            iov_len: buf.len(),
78        };
79        let cmsg_space = libc::CMSG_SPACE(std::mem::size_of::<RawFd>() as u32) as usize;
80        let mut cmsg_buf = vec![0u8; cmsg_space];
81        let mut msg: libc::msghdr = std::mem::zeroed();
82        msg.msg_iov = &mut iov;
83        msg.msg_iovlen = 1;
84        msg.msg_control = cmsg_buf.as_mut_ptr() as *mut libc::c_void;
85        msg.msg_controllen = cmsg_space as _;
86        let n = libc::recvmsg(channel, &mut msg, libc::MSG_DONTWAIT);
87        if n < 0 {
88            let err = std::io::Error::last_os_error();
89            if err.kind() == std::io::ErrorKind::WouldBlock {
90                return RecvFdResult::WouldBlock;
91            }
92            if err.raw_os_error() == Some(libc::EINTR) {
93                return RecvFdResult::WouldBlock;
94            }
95            return RecvFdResult::Closed;
96        }
97        if n == 0 {
98            return RecvFdResult::Closed;
99        }
100        let cmsg = libc::CMSG_FIRSTHDR(&msg);
101        if cmsg.is_null() {
102            return RecvFdResult::Closed;
103        }
104        if (*cmsg).cmsg_level == libc::SOL_SOCKET && (*cmsg).cmsg_type == libc::SCM_RIGHTS {
105            let fd_ptr = libc::CMSG_DATA(cmsg) as *const RawFd;
106            RecvFdResult::Fd(std::ptr::read_unaligned(fd_ptr))
107        } else {
108            RecvFdResult::Closed
109        }
110    }
111}
112
113pub async fn run_fd_channel(channel_fd: RawFd, state: crate::AppState) {
114    use std::os::unix::io::FromRawFd;
115    if state.0.verbose {
116        eprintln!("accepting clients via fd-channel (fd {channel_fd})");
117    }
118    let channel = unsafe { std::os::unix::net::UnixStream::from_raw_fd(channel_fd) };
119    channel.set_nonblocking(true).unwrap();
120    let async_channel = AsyncFd::new(channel).unwrap();
121    loop {
122        let mut guard = match async_channel.readable().await {
123            Ok(g) => g,
124            Err(e) => {
125                eprintln!("fd-channel error: {e}");
126                break;
127            }
128        };
129        match recv_fd(channel_fd) {
130            RecvFdResult::Fd(client_fd) => {
131                let std_stream = unsafe { std::os::unix::net::UnixStream::from_raw_fd(client_fd) };
132                std_stream.set_nonblocking(true).unwrap();
133                let stream = tokio::net::UnixStream::from_std(std_stream).unwrap();
134                let state = state.clone();
135                tokio::spawn(crate::handle_client(stream, state));
136                guard.retain_ready();
137            }
138            RecvFdResult::WouldBlock => {
139                guard.clear_ready();
140            }
141            RecvFdResult::Closed => {
142                break;
143            }
144        }
145    }
146    if state.0.verbose {
147        eprintln!("fd-channel closed, shutting down");
148    }
149}
blit_server/ipc/ipc_unix.rs

blit_server/ipc/
ipc_unix.rs
