lean_ctx/ipc/
mod.rs

1pub mod detached;
2pub mod process;
3pub mod runtime;
4
5#[cfg(unix)]
6mod unix;
7#[cfg(windows)]
8mod windows;
9#[cfg(windows)]
10pub use windows::NamedPipeListener;
11
12use std::path::PathBuf;
13
14use anyhow::Result;
15#[cfg(unix)]
16use anyhow::anyhow;
17#[cfg(unix)]
18use tokio::io::AsyncWriteExt;
19
20/// Directory of the daemon's endpoint files (`daemon.pid`, `daemon.sock`, the
21/// start lock and stderr log): the data directory, so `LEAN_CTX_DATA_DIR`, the
22/// XDG layout and a legacy `~/.lean-ctx` install each get their own daemon.
23/// The Windows pipe name is derived from it for the same reason.
24pub fn endpoint_dir() -> PathBuf {
25    crate::core::data_dir::lean_ctx_data_dir().unwrap_or_else(|_| legacy_endpoint_dir())
26}
27
28/// Where releases up to 3.12.0 kept the endpoint files: the OS data-local
29/// directory, whatever the data directory was. Read only to stop a daemon that
30/// such a release started.
31pub fn legacy_endpoint_dir() -> PathBuf {
32    dirs::data_local_dir()
33        .unwrap_or_else(|| dirs::home_dir().unwrap_or_default().join(".local/share"))
34        .join("lean-ctx")
35}
36
37/// Platform-independent daemon address.
38#[derive(Debug, Clone)]
39pub enum DaemonAddr {
40    #[cfg(unix)]
41    Unix(PathBuf),
42    #[cfg(windows)]
43    NamedPipe(String),
44}
45
46impl DaemonAddr {
47    pub fn default_for_current_os() -> Self {
48        #[cfg(unix)]
49        {
50            Self::Unix(unix::default_socket_path())
51        }
52        #[cfg(windows)]
53        {
54            Self::NamedPipe(windows::default_pipe_name())
55        }
56    }
57
58    pub fn display(&self) -> String {
59        match self {
60            #[cfg(unix)]
61            Self::Unix(p) => p.display().to_string(),
62            #[cfg(windows)]
63            Self::NamedPipe(n) => n.clone(),
64        }
65    }
66
67    /// Check whether anything is currently listening on this address.
68    pub fn is_listening(&self) -> bool {
69        match self {
70            #[cfg(unix)]
71            Self::Unix(p) => p.exists(),
72            #[cfg(windows)]
73            Self::NamedPipe(name) => windows::pipe_exists(name),
74        }
75    }
76}
77
78/// Remove any stale IPC endpoint (socket file / pipe marker).
79pub fn cleanup(addr: &DaemonAddr) {
80    match addr {
81        #[cfg(unix)]
82        DaemonAddr::Unix(p) => {
83            if p.exists() {
84                let _ = std::fs::remove_file(p);
85            }
86        }
87        #[cfg(windows)]
88        DaemonAddr::NamedPipe(_) => {
89            // Named pipes are kernel objects — no cleanup needed.
90        }
91    }
92}
93
94/// Bind a listener on the given address and return a platform-specific listener.
95#[cfg(unix)]
96pub fn bind_listener(addr: &DaemonAddr) -> Result<tokio::net::UnixListener> {
97    match addr {
98        DaemonAddr::Unix(path) => unix::bind_listener(path),
99    }
100}
101
102/// Connect to the daemon at the given address.
103#[cfg(unix)]
104pub async fn connect(addr: &DaemonAddr) -> Result<tokio::net::UnixStream> {
105    match addr {
106        DaemonAddr::Unix(path) => unix::connect(path).await,
107    }
108}
109
110/// Connect to a private runtime socket and send its versioned handshake.
111#[cfg(unix)]
112pub async fn connect_enterprise(
113    path: &std::path::Path,
114    handshake: &lean_ctx_protocol::runtime_handshake::RuntimeHandshake,
115) -> Result<tokio::net::UnixStream> {
116    let mut stream = unix::connect(path).await?;
117    let frame = handshake.encode_frame().map_err(|error| anyhow!(error))?;
118    stream.write_all(&frame).await?;
119    stream.flush().await?;
120    Ok(stream)
121}
122
123/// Bind a listener on the given Windows named pipe address.
124#[cfg(windows)]
125pub fn bind_listener(addr: &DaemonAddr) -> Result<NamedPipeListener> {
126    match addr {
127        DaemonAddr::NamedPipe(name) => NamedPipeListener::bind(name),
128    }
129}
130
131/// The security descriptor (SDDL) of every daemon pipe instance: owned by
132/// `sid`, with a protected DACL that grants `sid` alone full access. Without
133/// one a named pipe gets the default DACL, which also lets Everyone and
134/// Anonymous open it for reading; the daemon's routes take no token, so the
135/// pipe's ACL is their only guard, as the socket's 0600 mode is on Unix.
136#[cfg_attr(not(windows), allow(dead_code))]
137fn owner_only_pipe_sddl(sid: &str) -> String {
138    format!("O:{sid}D:P(A;;GA;;;{sid})")
139}
140
141/// Connect to the daemon at the given address.
142#[cfg(windows)]
143pub async fn connect(
144    addr: &DaemonAddr,
145) -> Result<tokio::net::windows::named_pipe::NamedPipeClient> {
146    match addr {
147        DaemonAddr::NamedPipe(name) => windows::connect(name).await,
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use super::*;
154
155    #[test]
156    fn daemon_addr_display_non_empty() {
157        let addr = DaemonAddr::default_for_current_os();
158        let display = addr.display();
159        assert!(!display.is_empty(), "`display` is empty");
160    }
161
162    // The daemon pipe used the default DACL (Everyone and Anonymous may read).
163    // Its descriptor names one owner, one access entry for that owner, and
164    // protects the DACL from inherited entries.
165    #[test]
166    fn the_daemon_pipe_descriptor_grants_its_owner_alone() {
167        let sid = "S-1-5-21-1-2-3-1001";
168        let sddl = owner_only_pipe_sddl(sid);
169        assert_eq!(sddl, format!("O:{sid}D:P(A;;GA;;;{sid})"));
170        assert_eq!(sddl.matches("(A;").count(), 1, "one access entry");
171        for broad in ["WD", "AN", "AU", "BU", "IU"] {
172            assert!(!sddl.contains(&format!(";;;{broad})")), "{broad}: {sddl}");
173        }
174    }
175
176    #[test]
177    fn cleanup_nonexistent_does_not_panic() {
178        #[cfg(unix)]
179        {
180            let addr = DaemonAddr::Unix(std::path::PathBuf::from(
181                "/tmp/lean-ctx-test-nonexistent.sock",
182            ));
183            cleanup(&addr);
184        }
185    }
186
187    #[cfg(unix)]
188    #[tokio::test]
189    async fn enterprise_connect_sends_versioned_frame() {
190        let dir = tempfile::tempdir().unwrap();
191        let path = dir.path().join("enterprise.sock");
192        let listener = unix::bind_listener(&path).unwrap();
193        let handshake = lean_ctx_protocol::runtime_handshake::RuntimeHandshake::public([]);
194        let client = connect_enterprise(&path, &handshake);
195        let server = listener.accept();
196        let (client, accepted) = tokio::join!(client, server);
197        let _client = client.unwrap();
198        let (mut stream, _) = accepted.unwrap();
199        use tokio::io::AsyncReadExt;
200        let mut length = [0_u8; 4];
201        stream.read_exact(&mut length).await.unwrap();
202        let len = u32::from_be_bytes(length) as usize;
203        assert_eq!(len, handshake.encode_frame().unwrap().len() - 4);
204    }
205}
lean_ctx/ipc/mod.rs

lean_ctx/ipc/
mod.rs
