greplm_core/

client.rs

//! Client for the greplm daemon.

#[cfg(unix)]
pub use unix_impl::Client;

#[cfg(not(unix))]
pub use stub_impl::Client;

// The daemon is built on Unix domain sockets and is unavailable on other
// platforms. This stub lets the CLI compile everywhere; `try_connect` always
// returns `None`, so callers transparently fall back to in-process queries.
#[cfg(not(unix))]
mod stub_impl {
    use std::path::Path;

    use crate::error::{Error, Result};
    use crate::proto::{Freshness, Request, Response};

    /// A connected client to a running greplm daemon (unsupported on this platform).
    pub struct Client {
        _private: (),
    }

    impl Client {
        /// Always returns `None`: no daemon is available on this platform.
        pub fn try_connect(_socket: &Path) -> Option<Client> {
            None
        }

        /// Always errors: no daemon is available on this platform.
        pub fn request(&mut self, _req: &Request) -> Result<Response> {
            Self::unsupported()
        }

        /// Always errors: no daemon is available on this platform.
        pub fn request_fresh(&mut self, _req: &Request, _freshness: Freshness) -> Result<Response> {
            Self::unsupported()
        }

        /// Always errors: no daemon is available on this platform.
        pub fn request_routed(
            &mut self,
            _root: &std::path::Path,
            _req: &Request,
        ) -> Result<Response> {
            Self::unsupported()
        }

        /// Always errors: no daemon is available on this platform.
        pub fn request_routed_fresh(
            &mut self,
            _root: &std::path::Path,
            _req: &Request,
            _freshness: Freshness,
        ) -> Result<Response> {
            Self::unsupported()
        }

        fn unsupported() -> Result<Response> {
            Err(Error::other(
                "greplm daemon is not supported on this platform",
            ))
        }
    }
}

#[cfg(unix)]
mod unix_impl {
    use std::io::{BufRead, BufReader, Read, Write};
    use std::os::unix::net::UnixStream;
    use std::path::Path;
    use std::time::Duration;

    use crate::error::{Error, Result};
    use crate::proto::{Freshness, LocalRequest, Request, Response, RoutedRequest};

    /// Default write timeout: a healthy daemon drains a request immediately.
    const WRITE_TIMEOUT: Duration = Duration::from_secs(10);
    /// Default read timeout: generous enough for a `strict`-freshness reindex on
    /// a large tree, but bounded so a wedged daemon can't hang the caller.
    /// Override with `GREPLM_DAEMON_TIMEOUT_MS`.
    const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(30);
    /// Cap on a single daemon response so a malicious or buggy peer can't stream
    /// unbounded bytes (no newline) and OOM the client. Generous enough for the
    /// largest legitimate `--exhaustive` result set.
    const MAX_RESPONSE_BYTES: u64 = 256 * 1024 * 1024;

    /// A connected client to a running greplm daemon.
    pub struct Client {
        reader: BufReader<UnixStream>,
        writer: UnixStream,
    }

    impl Client {
        /// Connect to a daemon listening on `socket`. Returns `None` if no daemon
        /// is reachable (so callers can fall back to in-process queries).
        pub fn try_connect(socket: &Path) -> Option<Client> {
            let stream = UnixStream::connect(socket).ok()?;
            // Defense in depth on top of the daemon's 0o600 socket in a 0o700
            // per-user dir: refuse to talk to a socket owned by another user, so
            // a squatter can't feed forged results to the agent. If the peer uid
            // can't be determined, proceed (the file-mode check still applies).
            if let (Some(peer), me) = (peer_uid(&stream), unsafe { libc::getuid() }) {
                if peer != me {
                    tracing::warn!(
                        "daemon socket {} is owned by uid {peer}, not {me}; ignoring it",
                        socket.display()
                    );
                    return None;
                }
            }
            // Bound both directions so a daemon that accepts the connection but
            // never replies (deadlocked, mid-reload, or a socket squatter) can't
            // hang the caller forever. On timeout the round-trip errors and the
            // caller falls back to the next transport — another daemon, or an
            // in-process query.
            let read_timeout = std::env::var("GREPLM_DAEMON_TIMEOUT_MS")
                .ok()
                .and_then(|s| s.parse::<u64>().ok())
                .filter(|&ms| ms > 0)
                .map(Duration::from_millis)
                .unwrap_or(DEFAULT_READ_TIMEOUT);
            let _ = stream.set_read_timeout(Some(read_timeout));
            let _ = stream.set_write_timeout(Some(WRITE_TIMEOUT));
            let reader = BufReader::new(stream.try_clone().ok()?);
            Some(Client {
                reader,
                writer: stream,
            })
        }

        /// Send a request to a per-project daemon and read the response (with
        /// default [`Freshness::Lazy`]).
        pub fn request(&mut self, req: &Request) -> Result<Response> {
            self.request_fresh(req, Freshness::Lazy)
        }

        /// Send a request to a per-project daemon at a chosen freshness level.
        pub fn request_fresh(&mut self, req: &Request, freshness: Freshness) -> Result<Response> {
            let env = LocalRequest {
                freshness,
                req: req.clone(),
            };
            self.round_trip(&env)
        }

        /// Send a request to the global multi-root daemon, addressed to a
        /// specific project `root` (with default [`Freshness::Lazy`]).
        pub fn request_routed(&mut self, root: &Path, req: &Request) -> Result<Response> {
            self.request_routed_fresh(root, req, Freshness::Lazy)
        }

        /// Send a request to the global multi-root daemon at a chosen freshness.
        pub fn request_routed_fresh(
            &mut self,
            root: &Path,
            req: &Request,
            freshness: Freshness,
        ) -> Result<Response> {
            let routed = RoutedRequest {
                root: root.to_path_buf(),
                freshness,
                req: req.clone(),
            };
            self.round_trip(&routed)
        }

        fn round_trip<T: serde::Serialize>(&mut self, value: &T) -> Result<Response> {
            let mut bytes = serde_json::to_vec(value)?;
            bytes.push(b'\n');
            self.writer.write_all(&bytes).map_err(Error::PlainIo)?;
            self.writer.flush().map_err(Error::PlainIo)?;
            let mut line = String::new();
            // Bound the response so a peer that streams bytes without a newline
            // can't grow `line` without limit.
            let n = (&mut self.reader)
                .take(MAX_RESPONSE_BYTES)
                .read_line(&mut line)
                .map_err(Error::PlainIo)?;
            if n == 0 {
                return Err(Error::other("daemon closed connection"));
            }
            if n as u64 >= MAX_RESPONSE_BYTES && !line.ends_with('\n') {
                return Err(Error::other("daemon response too large"));
            }
            let resp: Response = serde_json::from_str(line.trim())?;
            Ok(resp)
        }
    }

    /// The uid of the process on the other end of `stream`, or `None` if it
    /// can't be determined. Uses `SO_PEERCRED` on Linux and `getpeereid`
    /// elsewhere (macOS/BSD).
    fn peer_uid(stream: &UnixStream) -> Option<u32> {
        use std::os::unix::io::AsRawFd;
        let fd = stream.as_raw_fd();
        #[cfg(any(target_os = "linux", target_os = "android"))]
        {
            let mut cred = libc::ucred {
                pid: 0,
                uid: 0,
                gid: 0,
            };
            let mut len = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
            let rc = unsafe {
                libc::getsockopt(
                    fd,
                    libc::SOL_SOCKET,
                    libc::SO_PEERCRED,
                    &mut cred as *mut libc::ucred as *mut libc::c_void,
                    &mut len,
                )
            };
            (rc == 0).then_some(cred.uid)
        }
        #[cfg(not(any(target_os = "linux", target_os = "android")))]
        {
            let mut uid: libc::uid_t = 0;
            let mut gid: libc::gid_t = 0;
            let rc = unsafe { libc::getpeereid(fd, &mut uid, &mut gid) };
            (rc == 0).then_some(uid)
        }
    }

    #[cfg(test)]
    mod tests {
        use super::*;
        use std::os::unix::net::UnixListener;

        // The peer of a same-process connection is us — so the security check
        // reads the right uid and never false-rejects a same-user daemon.
        #[test]
        fn peer_uid_is_self_over_a_socket() {
            let dir = std::env::temp_dir().join(format!("greplm-peer-{}", std::process::id()));
            std::fs::create_dir_all(&dir).unwrap();
            let sock = dir.join("t.sock");
            let _ = std::fs::remove_file(&sock);
            let listener = UnixListener::bind(&sock).unwrap();
            let client = UnixStream::connect(&sock).unwrap();
            let (server, _) = listener.accept().unwrap();
            let me = unsafe { libc::getuid() };
            assert_eq!(peer_uid(&client), Some(me));
            assert_eq!(peer_uid(&server), Some(me));
            let _ = std::fs::remove_file(&sock);
            let _ = std::fs::remove_dir(&dir);
        }
    }
}