kache 1.2.2

//! Cross-platform local IPC transport for the daemon.
//!
//! The daemon's local socket and the wrapper / CLI clients use the
//! `interprocess` crate so the same API works on Unix domain sockets and
//! Windows named pipes. This module hides the `Name` resolution machinery
//! and provides a `prelude` with the trait imports callers need.
//!
//! # Naming
//!
//! kache addresses the daemon by a filesystem path (by default
//! `<runtime_dir>/daemon.sock`). On Unix that becomes the literal
//! UDS path. On Windows the path is translated to a named pipe under
//! `\\.\pipe\` by `interprocess`.

use anyhow::{Context, Result};
use std::path::Path;

#[cfg(test)]
pub use interprocess::local_socket::ListenerOptions;
pub use interprocess::local_socket::tokio::Stream as TokioStream;
pub use interprocess::local_socket::{Name, Stream as SyncStream};
// TokioListener is the type produced by `ListenerOptions::new().create_tokio()`.
// Re-exported so the daemon's `accept_loop` and its tests can name it explicitly.
pub use interprocess::local_socket::tokio::Listener as TokioListener;

/// Trait imports needed to call methods on the listener / stream types.
/// Most users want `use crate::transport::prelude::*;`.
pub mod prelude {
    pub use interprocess::local_socket::traits::tokio::{Listener as _, Stream as _};
    pub use interprocess::local_socket::traits::{Listener as _, Stream as _};
}

/// Why the daemon cannot bind a local socket at `path`, or `None` when it can.
///
/// A Unix socket address has a fixed-size path field, so a deep runtime
/// directory makes the daemon fail to start. kunobi-daemon reports the length
/// and the limit; this adds the two kache settings that move the socket.
/// Windows names a pipe after the path, which has no such limit.
pub fn socket_path_problem(path: &Path) -> Option<String> {
    #[cfg(unix)]
    if let Err(error) = kunobi_daemon::local::unix_socket::check_path(path) {
        return Some(format!(
            "{error}. Point KACHE_SOCKET_PATH at a shorter absolute path, or set a shorter \
             [cache] runtime_dir."
        ));
    }
    #[cfg(not(unix))]
    let _ = path;
    None
}

/// Bind through shared ownership and disable the transport's unchecked unlink.
pub async fn bind_daemon_listener(path: &Path) -> Result<Option<TokioListener>> {
    use interprocess::local_socket::traits::tokio::Listener as _;
    if let Some(problem) = socket_path_problem(path) {
        anyhow::bail!(problem);
    }
    #[cfg(unix)]
    let mut listener: TokioListener = {
        use kunobi_daemon::local::unix_socket::{self, Bound};
        let bound = unix_socket::acquire(path)
            .map_err(|error| anyhow::anyhow!("acquiring daemon socket: {error:?}"))?;
        let Bound::Won(listener) = bound else {
            return Ok(None);
        };
        let fd: std::os::fd::OwnedFd = listener.into();
        interprocess::os::unix::uds_local_socket::tokio::Listener::try_from(fd)?.into()
    };
    #[cfg(windows)]
    let mut listener = {
        use kunobi_daemon::local::windows_socket::{self, Bound};
        let bound = windows_socket::acquire_tokio(Path::new(&*daemon_endpoint(path)))
            .await
            .map_err(|error| anyhow::anyhow!("acquiring daemon pipe: {error:?}"))?;
        let Bound::Won(listener) = bound else {
            return Ok(None);
        };
        listener
    };
    listener.do_not_reclaim_name_on_drop();
    Ok(Some(listener))
}

/// The kunobi-daemon endpoint for a daemon socket path.
///
/// This name is where kache processes meet. A daemon and the clients looking
/// for it must derive the same value, so it is computed in exactly one place:
/// a second copy that drifted would leave a running daemon unreachable after
/// an upgrade. On Windows it is a hash of the path, because a named pipe lives
/// in its own namespace rather than on the filesystem.
pub fn daemon_endpoint(path: &Path) -> Box<kunobi_daemon::local::Endpoint> {
    #[cfg(unix)]
    {
        path.into()
    }
    #[cfg(windows)]
    {
        let hash = blake3::hash(path.as_os_str().as_encoded_bytes());
        format!("kache-daemon-{}", &hash.to_hex()[..16]).into_boxed_str()
    }
}

/// Build the platform-appropriate IPC name for a path.
///
/// On Unix, the path is used directly as a filesystem UDS path via
/// `GenericFilePath`. On Windows, the path cannot be used as a named pipe
/// path directly (named pipes must live under `\\.\pipe\`), so we derive a
/// namespaced name from the path via `GenericNamespaced`.
pub fn socket_name(path: &Path) -> Result<Name<'static>> {
    #[cfg(unix)]
    {
        use interprocess::local_socket::{GenericFilePath, ToFsName};
        path.to_fs_name::<GenericFilePath>()
            .with_context(|| format!("converting {} to local-socket name", path.display()))
            .map(|n| n.into_owned())
    }
    #[cfg(windows)]
    {
        use interprocess::local_socket::{GenericNamespaced, ToNsName};
        daemon_endpoint(path)
            .to_ns_name::<GenericNamespaced>()
            .with_context(|| format!("converting {} to named pipe name", path.display()))
            .map(|n| n.into_owned())
    }
}

/// Reject a connection whose peer is not the daemon's own user.
///
/// Defense-in-depth behind the `0600` mode kunobi-daemon gives the socket: even
/// if the socket file ends up reachable by others (nonstandard umask, relinked
/// path, shared cache dir), every accepted connection is checked against the peer's
/// effective UID via `SO_PEERCRED` (Linux) / `getpeereid` (macOS, BSDs).
/// Fails closed when credentials cannot be determined — a peer we cannot
/// authenticate is a peer we refuse. Windows named pipes carry peer identity
/// in the kernel object ACL rather than creds messages, so the check is
/// Unix-only.
///
/// The decision is taken over an `Option<u32>` rather than a live stream so
/// every branch is assertable: CI cannot open a connection from a foreign
/// UID, and cannot make the platform withhold credentials.
#[cfg(unix)]
pub fn require_self_peer(peer_euid: Option<u32>) -> std::io::Result<()> {
    match peer_euid {
        Some(peer_uid) if peer_uid_is_self(peer_uid) => Ok(()),
        Some(peer_uid) => Err(std::io::Error::new(
            std::io::ErrorKind::PermissionDenied,
            format!(
                "peer uid {peer_uid} does not match daemon uid {}",
                self_uid()
            ),
        )),
        None => Err(std::io::Error::new(
            std::io::ErrorKind::PermissionDenied,
            format!("peer credentials unavailable (daemon uid {})", self_uid()),
        )),
    }
}

/// The peer's effective UID, where the platform exposes it: `SO_PEERCRED` on
/// Linux, `getpeereid` on macOS and the BSDs. `None` means we could not
/// authenticate the peer, which [`require_self_peer`] treats as a refusal.
#[cfg(unix)]
pub fn peer_euid(stream: &TokioStream) -> Option<u32> {
    use interprocess::local_socket::traits::StreamCommon as _;
    stream.peer_creds().ok().and_then(|creds| creds.euid())
}

/// The daemon's effective UID (test seam: the comparison lives here so the
/// accept/reject decision is assertable without a foreign-UID connection).
#[cfg(unix)]
fn self_uid() -> u32 {
    unsafe { libc::geteuid() }
}

/// Does a peer's effective UID belong to the daemon's own user?
#[cfg(unix)]
fn peer_uid_is_self(peer_uid: u32) -> bool {
    peer_uid == self_uid()
}

/// True if a daemon socket / named pipe at `path` accepts a connection
/// right now. Used by liveness probes — does not check whether the daemon
/// is responsive, only whether *something* is listening.
pub fn is_reachable(path: &Path) -> bool {
    use interprocess::local_socket::traits::Stream as _;
    let Ok(name) = socket_name(path) else {
        return false;
    };
    SyncStream::connect(name).is_ok()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[cfg(unix)]
    #[test]
    fn a_socket_path_fits_up_to_the_platform_limit_and_not_one_byte_more() {
        let max = kunobi_daemon::local::unix_socket::MAX_PATH_BYTES;
        let fits = format!("/{}", "a".repeat(max - 1));
        assert_eq!(fits.len(), max);
        assert_eq!(socket_path_problem(Path::new(&fits)), None);

        let too_long = format!("{fits}b");
        let problem = socket_path_problem(Path::new(&too_long)).expect("one byte over the limit");
        assert!(problem.contains(&format!("{} bytes", max + 1)), "{problem}");
        assert!(problem.contains("KACHE_SOCKET_PATH"), "{problem}");
        assert!(problem.contains("runtime_dir"), "{problem}");
    }

    #[cfg(unix)]
    #[test]
    fn binding_an_overlong_path_names_the_limit() {
        let path = format!(
            "/{}",
            "a".repeat(kunobi_daemon::local::unix_socket::MAX_PATH_BYTES)
        );
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let Err(error) = runtime.block_on(bind_daemon_listener(Path::new(&path))) else {
            panic!("an overlong path cannot bind");
        };
        assert!(error.to_string().contains("KACHE_SOCKET_PATH"), "{error:#}");
    }

    #[test]
    fn socket_name_builds_for_a_plausible_path() {
        // On Unix the path becomes a UDS filesystem name; on Windows it is
        // hashed into a `\\.\pipe\` namespaced name. Either way a normal path
        // must resolve without error.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("kache-daemon.sock");
        assert!(socket_name(&path).is_ok());
    }

    #[test]
    fn is_reachable_is_false_when_nothing_listens() {
        // A path under a fresh temp dir has no listener bound, so a connect
        // attempt must fail and `is_reachable` must report false.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("definitely-not-listening.sock");
        assert!(!is_reachable(&path));
    }

    #[cfg(unix)]
    #[test]
    fn is_reachable_is_true_for_a_live_listener() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("live.sock");
        let name = socket_name(&path).unwrap();
        // Hold the listener open for the duration of the probe.
        let _listener = ListenerOptions::new()
            .name(name)
            .create_sync()
            .expect("bind listener");
        assert!(is_reachable(&path));
    }

    #[cfg(unix)]
    #[test]
    fn require_self_peer_rejects_foreign_and_unauthenticated_peers() {
        // The accept direction is also covered end-to-end by
        // `same_user_peer_check_accepts_own_connections`; this pins every
        // branch, including the two CI cannot produce with a real
        // connection: a foreign UID, and a platform that withholds creds.
        let mine = self_uid();
        assert!(require_self_peer(Some(mine)).is_ok());

        let foreign = require_self_peer(Some(mine.wrapping_add(1))).unwrap_err();
        assert_eq!(foreign.kind(), std::io::ErrorKind::PermissionDenied);
        assert!(foreign.to_string().contains("does not match daemon uid"));

        // Fail closed: a peer we cannot authenticate is a peer we refuse.
        let unknown = require_self_peer(None).unwrap_err();
        assert_eq!(unknown.kind(), std::io::ErrorKind::PermissionDenied);
        assert!(unknown.to_string().contains("credentials unavailable"));

        assert!(peer_uid_is_self(mine));
        assert!(!peer_uid_is_self(mine.wrapping_add(0xdead_beef)));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn same_user_peer_check_accepts_own_connections() {
        use crate::transport::prelude::*;
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("creds.sock");
        let name = socket_name(&path).unwrap();
        let listener = ListenerOptions::new()
            .name(name)
            .create_tokio()
            .expect("bind listener");

        let client = tokio::spawn(async move {
            TokioStream::connect(socket_name(&path).unwrap())
                .await
                .expect("connect")
        });
        let (stream, _client) = tokio::join!(listener.accept(), client);

        // The test client runs under the same effective UID as the "daemon",
        // so the credential gate must accept it. Rejection of foreign UIDs is
        // not unit-testable without privilege switching.
        let stream = stream.expect("accept");
        require_self_peer(peer_euid(&stream)).expect("same-user peer accepted");
    }
}