hotl_platform/ipc/
mod.rs

1//! [`Ipc`] — a private, same-user, local-only session endpoint.
2
3use std::future::Future;
4use std::io;
5use std::path::PathBuf;
6
7#[cfg(unix)]
8mod unix;
9#[cfg(unix)]
10pub use unix::{UnixIpc, UnixIpcListener, UnixIpcStream};
11#[cfg(unix)]
12pub type ActiveIpc = UnixIpc;
13
14#[cfg(windows)]
15mod windows;
16#[cfg(windows)]
17pub use windows::{WindowsIpc, WindowsIpcListener, WindowsIpcStream};
18#[cfg(windows)]
19pub type ActiveIpc = WindowsIpc;
20
21/// A private, same-user, local-only session endpoint.
22///
23/// CONTRACT: [`bind_private`](Ipc::bind_private) returns an endpoint reachable
24/// **only** by the current user, and that restriction is applied at bind, not
25/// after. The authorization boundary is the OS object's own access control —
26/// the `0600` mode on Unix, the DACL on Windows.
27/// [`authenticate_peer`](Ipc::authenticate_peer) is defence in depth and is
28/// never the only check.
29pub trait Ipc: crate::sealed::Sealed {
30    type Listener: IpcListener<Stream = Self::Stream>;
31    type Stream: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static;
32
33    /// Whether a dead server leaves an artifact that must be swept.
34    ///
35    /// Unix: `true` — a socket file, hence the caller's unlink guard. Windows:
36    /// `false`, because a named pipe is a kernel object that vanishes with its
37    /// last handle. The const says so, so the guard is a documented no-op
38    /// rather than dead machinery ported for symmetry.
39    const LEAVES_STALE_ARTIFACT: bool;
40
41    fn bind_private(&self, id: &str) -> io::Result<Self::Listener>;
42    fn connect(&self, id: &str) -> impl Future<Output = io::Result<Self::Stream>> + Send;
43
44    /// Reject a peer that is not the current user.
45    ///
46    /// Unix: `SO_PEERCRED` via `peer_cred()`. Windows:
47    /// `ImpersonateNamedPipeClient` + `EqualSid` + `RevertToSelf`, which is the
48    /// correct analogue because it captures the client's token *as of the
49    /// connection*.
50    ///
51    /// **Not `GetNamedPipeClientProcessId` + `OpenProcess`.** That is the
52    /// obvious-looking answer and it is racy — the pid can be reused between
53    /// the connect and the lookup, where `SO_PEERCRED` cannot. Written down
54    /// because someone will propose it.
55    fn authenticate_peer(&self, stream: &Self::Stream) -> Result<(), PeerReject>;
56
57    /// Is a server listening on `id`?
58    ///
59    /// Tri-state on purpose. A two-state bool is how the Windows port ships a
60    /// bug: `ERROR_PIPE_BUSY` means a server **exists** and all its instances
61    /// are busy — i.e. LIVE. Reading it as dead makes hotl steal a running
62    /// session's pipe name.
63    fn liveness(&self, id: &str) -> Liveness;
64
65    fn list_live(&self) -> Vec<String>;
66
67    /// The on-disk artifact for `id`, when the platform has one. `None` on
68    /// Windows, where there is nothing to unlink.
69    fn artifact_path(&self, id: &str) -> Option<PathBuf>;
70}
71
72/// Accepting is a method on the listener rather than the adapter, because a
73/// named-pipe server must create the *next* instance as part of accepting the
74/// current one — there is state to carry that a stateless call cannot.
75pub trait IpcListener: Send {
76    type Stream;
77    fn accept(&mut self) -> impl Future<Output = io::Result<Self::Stream>> + Send;
78}
79
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum Liveness {
82    Live,
83    Dead,
84}
85
86#[derive(Debug)]
87pub struct PeerReject(pub String);
88
89impl std::fmt::Display for PeerReject {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        write!(f, "{}", self.0)
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
99
100    /// One body, both transports: bind, connect, authenticate, round-trip a
101    /// frame, and see the endpoint go from live to dead.
102    #[test]
103    fn active_adapter_upholds_the_contract() {
104        let rt = tokio::runtime::Builder::new_multi_thread()
105            .enable_all()
106            .build()
107            .unwrap();
108        rt.block_on(async {
109            let ipc = crate::IPC;
110            let id = format!("hotl-ipc-test-{}", std::process::id());
111            assert_eq!(ipc.liveness(&id), Liveness::Dead, "nothing bound yet");
112
113            let mut listener = ipc.bind_private(&id).unwrap();
114            assert_eq!(ipc.liveness(&id), Liveness::Live);
115            assert!(ipc.list_live().contains(&id), "a bound endpoint is listed");
116
117            // A real accept loop tolerates a peer that connects and vanishes,
118            // and it must here: the two `liveness` probes above each left a
119            // dead connection queued in the backlog ahead of the real client.
120            //
121            // Authentication alone cannot be the filter, and that is a platform
122            // difference worth stating: macOS's `getpeereid` *fails* on a peer
123            // that already closed, so a dead connection is rejected there by
124            // accident. Linux's `SO_PEERCRED` still reports the credentials of
125            // the peer that connected, so the same dead connection
126            // authenticates fine and is only distinguishable by the fact that
127            // it never says anything. Discard on either signal and keep
128            // accepting — which is exactly what `session_server::authenticate`
129            // does with its bounded handshake read.
130            let server = tokio::spawn(async move {
131                loop {
132                    let s = listener.accept().await.unwrap();
133                    if crate::IPC.authenticate_peer(&s).is_err() {
134                        continue;
135                    }
136                    let (r, mut w) = tokio::io::split(s);
137                    let mut lines = BufReader::new(r).lines();
138                    let Ok(Some(got)) = lines.next_line().await else {
139                        continue; // connected and vanished — a probe, not a client
140                    };
141                    w.write_all(format!("echo:{got}\n").as_bytes())
142                        .await
143                        .unwrap();
144                    w.flush().await.unwrap();
145                    return;
146                }
147            });
148
149            let client = ipc.connect(&id).await.unwrap();
150            let (r, mut w) = tokio::io::split(client);
151            w.write_all(b"ping\n").await.unwrap();
152            w.flush().await.unwrap();
153            let mut lines = BufReader::new(r).lines();
154            assert_eq!(lines.next_line().await.unwrap().unwrap(), "echo:ping");
155            server.await.unwrap();
156
157            // An adapter that names an artifact must admit it leaves one, and
158            // vice versa — the const and the path have to agree or a caller's
159            // sweep is either dead code or a missing cleanup.
160            assert_eq!(
161                ipc.artifact_path(&id).is_some(),
162                <ActiveIpc as Ipc>::LEAVES_STALE_ARTIFACT
163            );
164            if let Some(p) = ipc.artifact_path(&id) {
165                let _ = std::fs::remove_file(p);
166            }
167        });
168    }
169}
hotl_platform/ipc/mod.rs

hotl_platform/ipc/
mod.rs
