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

blit_server/ipc/
ipc_unix.rs
