diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8ad67a7..a6737a2 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -133,6 +133,31 @@ The supervisor binds sockets once at cold start and passes them to every child v The kernel-level socket (and its accept queue) is never closed. Connections arriving during the O→N transition queue in the kernel and N's first `accept()` picks them up. The listen socket never goes "down" from a client's perspective. +**Adoption is validated, not assumed.** `LISTEN_FDS`/`LISTEN_FDNAMES`/`LISTEN_PID` are systemd's variables, so they can reach a process from something other than a handoff supervisor, and `FromRawFd` checks nothing. `role.rs` therefore: + +1. Ignores the whole block when `LISTEN_PID` is present and names another process (systemd's activation contract). This supervisor cannot set `LISTEN_PID` — the child's pid is only knowable after `fork`, and `Command`'s environment is materialized before `pre_exec` — so its *absence* is not suspicious; successor identity is verified on the wire by the `Hello` pid check instead. +2. Skips any advertised slot that `fstat` does not report as a socket, and requires the listening state plus the right address family (`InheritedListeners::take` for TCP, `take_unix` for `AF_UNIX`) before handing back a typed listener. A refused slot is left untouched rather than wrapped — a wrapper would close a descriptor belonging to something else when it drops. + +**Adopted descriptors are normalized.** `FD_CLOEXEC` is re-armed (the parent's `dup2` cleared it so the FD would survive `execve`; leaving it clear leaks listeners and the control socket into every subprocess the daemon later spawns — and a leaked control-socket copy holds the supervisor's EOF open, hiding the primitive's death) and `O_NONBLOCK` is cleared (it lives on the open file description, so a parent that gave its listener to an async runtime would otherwise hand the child a listener whose first `accept()` returns `EAGAIN`). + +### Control socket ownership and access control + +A control connection can drain and seal the daemon, so the socket is a privileged surface, not just an IPC detail: + +- **Bound `0600`, atomically.** `sock::bind_socket` binds a short staging name in the target directory, chmods it, then `rename(2)`s it over the published path. The socket is never reachable while its mode is still `0777 & ~umask`, and a rebind never leaves a window where the path is missing (which the previous unlink-then-bind sequence did — a client connecting inside it gets `ENOENT`, and a second binder racing in it silently steals the name). +- **Peer-uid checked.** Every accepted connection's uid is read from the kernel (`SO_PEERCRED` on Linux, `getpeereid(3)` elsewhere) — latched at `connect(2)`, so it cannot be forged or raced. The daemon's own euid and root are accepted; everything else is refused. Mode bits alone are not sufficient: they can be loosened by an operator, and some filesystems ignore them. +- **Path length validated up front.** `Supervisor::new` and every bind reject a path that does not fit in `sockaddr_un.sun_path` (108 bytes on Linux, 104 elsewhere) with `Error::SocketPathTooLong`, instead of surfacing an opaque `EINVAL` mid-handoff or, on some BSDs, binding a silently truncated path. + +### Accept-loop resilience + +`Incumbent::serve` distinguishes "this *connection* failed" from "this *listener* is broken", because leaving the loop is far more expensive than it looks: the process keeps serving traffic but stops answering the control socket, so it can never be handed off again and the only way to deploy a new build is a hard restart — the downtime this crate exists to avoid. + +| `accept(2)` error | Response | +|---|---| +| `EINTR`, `ECONNABORTED`, `EPROTO`, `EPERM` | Retry immediately (the peer went away, or a firewall hook rejected it) | +| `EMFILE`, `ENFILE`, `ENOBUFS`, `ENOMEM`, `EAGAIN` | Sleep and retry, exponential 5 ms → 1 s, reset on the next success; the connection stays queued in the backlog | +| anything else (`EBADF`, `EINVAL`, `ENOTSOCK`, …) | Fatal — the listener is unusable and retrying would spin | + ### Flock ordering (the load-bearing piece) O releases the flock in `SealRequest` handling (`incumbent.rs:run_session_loop`), immediately after `drainable.seal()` succeeds — before sending `SealComplete`, before receiving `Commit`, before exiting. This is the critical ordering: @@ -157,6 +182,10 @@ Every message is a length-prefixed frame over a `UnixStream`: Frame size is capped at 1 MiB (`MAX_FRAME_BYTES`) to bound allocation on the reader side. The `Message` enum's variant discriminant is encoded by postcard as part of the payload — no separate type byte. See `frame.rs`. +**Reads are resumable.** Every protocol read is armed with `SO_RCVTIMEO`, and `read_exact` discards what it already consumed when a read fails — so a frame straddling a timeout boundary leaves the stream mid-frame and the next read interprets payload bytes as a length prefix, corrupting every frame after it. `FrameAccumulator` owns the partial bytes instead: a timeout is a *suspension* (`Ok(None)`), and the caller decides whether to keep waiting or give up. Callers use `has_partial()` to tell "peer has gone silent" (nothing buffered — treat as peer-dead) from "peer is mid-frame" (buffered bytes prove liveness; only the wall-clock budget applies). + +**Writes are bounded and signal-safe.** Control-socket frames go out through `write_frame` → `sock::send_all`, which uses `MSG_NOSIGNAL` (Linux) or `SO_NOSIGPIPE` (macOS/BSD): a peer that died mid-handoff yields `EPIPE`, never a process-killing `SIGPIPE`. Rust's runtime ignores `SIGPIPE` by default, but that is a property of the embedding *binary*, not of this library. Endpoints also carry `SO_SNDTIMEO` (`CONTROL_WRITE_TIMEOUT`, 10 s), since a peer that stops reading would otherwise park the writer in an unbounded `write(2)` — the one place in the handoff where neither the liveness clock nor the deadline is running, both being enforced on the read side. + ### Protocol negotiation Both sides announce `proto_min`/`proto_max` in `Hello`. `negotiate_version()` picks the highest version in the intersection. If ranges are disjoint, returns `Error::VersionMismatch` and the connection is closed before any handoff begins. Currently only version 1 exists (`PROTO_MIN == PROTO_MAX == 1`). @@ -172,7 +201,7 @@ Journal writes use `postcard` serialization; the rename makes each write crash-s `DataDirLock::acquire_or_break_stale()` handles the case where the lockfile holds a PID that is no longer alive: 1. Try `acquire()`. If it succeeds, done. -2. If `LockHeld`, read the pidfile and call `kill(pid, 0)`. +2. If `LockHeld`, read the pidfile and call `kill(pid, 0)`. Only `ESRCH` counts as dead: `EPERM` means the process exists but runs as another user (a daemon restarted under a different service account, or an operator recovering as non-root), and reading that as "dead" would send the stale-break path after a live holder. 3. If the PID is alive: return `StaleLockBreakRefused`. Never break a live holder. 4. If the PID is dead: retry `acquire()` on the same lockfile inode. The kernel released the flock when the holder process died, so the second attempt succeeds. If something is still genuinely holding the flock (an inherited FD outliving the named holder, or a brief PID-reuse race), the retry returns `LockHeld` again and we surface `StaleLockBreakRefused`. @@ -180,6 +209,8 @@ We deliberately do **not** unlink the lockfile and acquire on a fresh inode: tha See `lock.rs:acquire_or_break_stale()`. +**Filesystem requirement.** `flock(2)` excludes only the contenders a single kernel sees. Local filesystems (ext4, xfs, btrfs, zfs, apfs, ufs) are the supported configuration; NFS, CIFS/SMB, 9p, and FUSE without lock forwarding either emulate `flock` with different (per-process) semantics or degrade to a purely local lock that two hosts will both "acquire". Two containers bind-mounting one host directory are fine (same kernel); two hosts sharing network storage are not. A lock that does not exclude produces exactly the two-writer failure this crate exists to prevent, and the holder cannot detect it — put the data directory on a local filesystem. + ## State Machine ### Journal phase machine (supervisor-side) @@ -294,6 +325,9 @@ When O's `serve()` loop catches a session error and the data-dir flock is not he - `handoff_id` in every message matches the active handoff — rejects replayed or cross-session messages. - Frame length is ≤ 1 MiB — prevents unbounded allocation from a malformed peer. - `Commit` is not accepted before `SealComplete` — protocol ordering is enforced. +- The connecting peer's uid (from `SO_PEERCRED`/`getpeereid`) is the daemon's own or root — on both the control socket and the reference supervisor's trigger socket. +- Inherited listener descriptors are open sockets of the expected family and listening state, and `LISTEN_PID`, when present, names this process. +- A binary named by a trigger client is on the configured allowlist. **What passes through unchecked:** @@ -304,7 +338,9 @@ When O's `serve()` loop catches a session error and the data-dir flock is not he **Why these boundaries are where they are:** -The library is embedded in a trusted, same-host supervisor process. All three roles (S, O, N) are spawned by the same operator; no external network is involved. Authentication between them is not needed — the Unix socket path is the security boundary. If an untrusted process can connect to the control socket, the host is already compromised. +The library is embedded in a trusted, same-host supervisor process. All three roles (S, O, N) are spawned by the same operator and no external network is involved, so the checks stop at *which local user* is talking — there is no cryptographic authentication of message contents. + +The socket path alone is not treated as the boundary, though. Filesystem permissions on a Unix socket depend on the process umask and are ignored by some filesystems, and a multi-tenant host routinely has unprivileged local users who should not be able to drain a database. So the sockets are bound `0600` *and* every peer's uid is checked; the reference supervisor additionally refuses to exec a binary a client names unless it is on an allowlist. These are cheap, and they change "any local uid can force a drain+seal" into "a local uid that is already the service account or root can". ## Package Structure @@ -312,7 +348,8 @@ The library is embedded in a trusted, same-host supervisor process. All three ro |------|-------------| | `crates/handoff/src/lib.rs` | Re-exports public surface; no logic | | `crates/handoff/src/protocol.rs` | Wire message enum + framing constants; `negotiate_version()` | -| `crates/handoff/src/frame.rs` | `read_message` / `write_message`; 1 MiB frame cap | +| `crates/handoff/src/frame.rs` | `read_message` / `write_message` / `write_frame`; `FrameAccumulator` (resumable reads); 1 MiB frame cap | +| `crates/handoff/src/sock.rs` | Unix-socket mechanics: `sun_path` validation, atomic `0600` bind, peer-uid lookup, FD flag normalization, SIGPIPE-safe writes | | `crates/handoff/src/supervisor.rs` | `Supervisor::perform_handoff()`; `ChildGuard`; journal writes; `spawn_successor()` pre_exec dance | | `crates/handoff/src/incumbent.rs` | `Incumbent::serve()`; per-session state machine; flock release on seal | | `crates/handoff/src/drainable.rs` | `Drainable` trait + report types (`DrainReport`, `SealReport`, `StateSnapshot`) | @@ -334,7 +371,9 @@ The library is embedded in a trusted, same-host supervisor process. All three ro | `args` | `[]` | Argv passed to every primitive spawn | | `env` | `[]` | Extra env vars merged into every spawn's environment | | `listeners` | `[]` | Sockets S binds at startup; inherited by every primitive via LISTEN_FDS | -| `trigger_socket` | (required) | Unix socket S listens on for `handoff [binary]` trigger commands | +| `trigger_socket` | (required) | Unix socket S listens on for `handoff [binary]` trigger commands; bound `0600`, peer-uid checked, 10 s I/O timeout and 4 KiB command cap per client | +| `allowed_uids` | `[]` | Extra uids permitted to issue trigger commands (S's own uid and root always are) | +| `allowed_binaries` | `[]` | Paths a trigger client may name in `handoff `, in addition to `binary`. Compared after canonicalization; anything else is refused — an unrestricted override is a "run this file as the supervisor's user" primitive | | `journal` | `None` | If set, S writes phase journal here for crash recovery | | `drain_grace_secs` | `25` | Budget S gives O for the drain phase before sending `Drained`; S's read for the reply extends `WIRE_SLACK` (1 s) past this cap | | `deadline_secs` | `60` | Overall handoff deadline (post-drain through Ready); S's reads for `SealComplete` and `Ready` extend `WIRE_SLACK` (1 s) past this cap | @@ -359,7 +398,10 @@ The library is embedded in a trusted, same-host supervisor process. All three ro | Second `PrepareHandoff` with different `handoff_id` | O returns `Error::HandoffInProgress`; session closes | S observes disconnect; must start fresh session | | Frame > 1 MiB received | `Error::FrameTooLarge`; connection closed | Peer has a bug; reconnect | | Protocol version mismatch | `Error::VersionMismatch`; connection closed before any handoff begins | Upgrade either S or the primitive | -| Stale pidfile, holder dead | `acquire_or_break_stale()` detects dead PID via `kill(pid, 0)`; unlinks files; acquires on fresh inode | Automatic; no manual intervention | +| Stale pidfile, holder dead | `acquire_or_break_stale()` detects `ESRCH` from `kill(pid, 0)`; retries `acquire()` on the same lockfile inode | Automatic; no manual intervention | +| `accept()` fails transiently (`ECONNABORTED`, `EMFILE`, …) | O retries, backing off 5 ms → 1 s on resource exhaustion; the serve loop survives | Automatic; the queued connection is picked up when capacity returns | +| Control frame straddles a receive timeout | `FrameAccumulator` retains the partial frame; the read resumes where it stopped | Automatic; no stream desynchronization | +| Control peer is an unauthorized local uid | Connection refused before any protocol frame is read; logged at WARN | Add the uid to `allowed_uids` if it is legitimate | | Stale pidfile, holder alive | `Error::StaleLockBreakRefused` — refuses to evict a live process | Manual investigation required; indicates two supervisors for one data dir | ## Observability @@ -421,7 +463,10 @@ impl HandshookSuccessor { pub struct BegunSuccessor; impl BegunSuccessor { + /// `None` if the name was not passed, was already taken, or the + /// descriptor is not a listening socket of the expected family. pub fn take_listener(&mut self, name: &str) -> Option; + pub fn take_unix_listener(&mut self, name: &str) -> Option; pub fn handoff_id(&self) -> HandoffId; pub fn listener_names(&self) -> Vec; /// Send `Ready`. Caller must NOT bind the control socket until the @@ -440,9 +485,10 @@ impl BegunSuccessor { pub struct Incumbent; impl Incumbent { - /// Cold-start bind. Unlinks any stale socket file before binding; - /// safe only when no prior incumbent is alive on this path. From a - /// successor, use `Successor::announce_and_bind` instead. + /// Cold-start bind. Takes over the path from any stale binding via + /// bind-then-rename (mode `0600`, no missing-path window); safe only + /// when no prior incumbent is alive on this path. From a successor, + /// use `Successor::announce_and_bind` instead. pub fn bind_cold_start(socket_path: &Path, lock: DataDirLock) -> Result; pub fn with_build_id(self, build_id: Vec) -> Self; pub fn serve(self, drainable: D) -> Result<()>; diff --git a/Cargo.lock b/Cargo.lock index be68a13..493d5b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -63,9 +63,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "atomic-polyfill" @@ -834,9 +834,9 @@ dependencies = [ [[package]] name = "spin" -version = "0.9.8" +version = "0.9.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" dependencies = [ "lock_api", ] diff --git a/README.md b/README.md index e5636ec..552dedc 100644 --- a/README.md +++ b/README.md @@ -142,6 +142,12 @@ binary = "/usr/local/bin/my-daemon" drain_grace_secs = 25 deadline_secs = 60 +# The trigger socket is bound 0600 and only accepts connections from the +# supervisor's own uid or root; list any other uid that may trigger swaps. +allowed_uids = [] +# A trigger client may only name a binary listed here (or `binary` above). +allowed_binaries = ["/usr/local/bin/my-daemon-v2"] + [[listeners]] name = "http" addr = "0.0.0.0:8080" @@ -151,9 +157,12 @@ addr = "0.0.0.0:8080" # Start the supervisor (it cold-starts your daemon): handoff-supervisor --config handoff.toml -# Later, trigger a swap to a new binary: -echo "handoff /usr/local/bin/my-daemon-v2" | socat - UNIX-CONNECT:/run/my-daemon/handoff.trigger +# Later, swap to a new build of the configured binary: +echo "handoff" | socat - UNIX-CONNECT:/run/my-daemon/handoff.trigger # ok: handoff_id=... committed=true abort_reason=None + +# Or name a binary from `allowed_binaries`; anything else is refused: +echo "handoff /usr/local/bin/my-daemon-v2" | socat - UNIX-CONNECT:/run/my-daemon/handoff.trigger ``` ### Embedded in code diff --git a/crates/handoff-supervisor/src/main.rs b/crates/handoff-supervisor/src/main.rs index f1090db..1fecf67 100644 --- a/crates/handoff-supervisor/src/main.rs +++ b/crates/handoff-supervisor/src/main.rs @@ -10,6 +10,11 @@ //! drive `handoff::Supervisor::perform_handoff` against the running //! primitive's control socket. //! +//! The trigger socket is a privileged control surface: a command on it makes +//! the supervisor drain the daemon and exec a binary. It is bound `0600` and +//! every client's peer uid is checked (see `allowed_uids`), and the binary a +//! client may name is restricted to `binary` plus `allowed_binaries`. +//! //! This binary is a demonstration of the library API and a convenience for //! local development and tests. Production embedders (`guest-agent`, //! `beyond-pg`) link `handoff` directly and integrate it with their own @@ -17,11 +22,11 @@ #![deny(unsafe_code)] -use std::io::{BufRead, BufReader, Write}; +use std::io::{BufRead, BufReader, Read, Write}; use std::net::TcpListener; use std::os::fd::{AsRawFd, RawFd}; -use std::os::unix::net::UnixListener; -use std::path::PathBuf; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -31,8 +36,18 @@ use clap::Parser; use serde::Deserialize; use tracing_subscriber::EnvFilter; -use handoff::pass_listener_fds_on_spawn; use handoff::supervisor::{SpawnSpec, Supervisor}; +use handoff::{bind_socket, pass_listener_fds_on_spawn, peer_uid, peer_uid_is_allowed}; + +/// Cap on how long a trigger client may take to send its command line, and +/// on how long a reply write may block. The loop is single-threaded, so an +/// unbounded read here is a denial of service against every future handoff: +/// one client that connects and never writes wedges the supervisor forever. +const TRIGGER_IO_TIMEOUT: Duration = Duration::from_secs(10); +/// Cap on a trigger command line. Commands are a verb plus an optional path; +/// anything longer is a client sending garbage, and reading it unbounded +/// would let one connection grow the supervisor's memory without limit. +const TRIGGER_MAX_LINE: u64 = 4096; #[derive(Parser, Debug)] #[command( @@ -60,6 +75,18 @@ struct Config { listeners: Vec, /// Local Unix socket the supervisor listens on for trigger commands. trigger_socket: PathBuf, + /// Additional uids allowed to issue trigger commands. The supervisor's + /// own uid and root are always allowed; every other peer is refused. + /// The socket is bound `0600`, so this only widens access when a + /// deployment deliberately relaxes the directory permissions too. + #[serde(default)] + allowed_uids: Vec, + /// Binaries a trigger client may name in `handoff `, in addition + /// to `binary`. A client-supplied path that is not in this set is + /// refused: the trigger would otherwise be a request to execute an + /// arbitrary file as the supervisor's user. + #[serde(default)] + allowed_binaries: Vec, #[serde(default)] journal: Option, #[serde(default = "default_drain_grace_secs")] @@ -134,12 +161,10 @@ fn main() -> Result<()> { Err(e) => tracing::warn!(error = %e, "resume_from_journal failed; continuing"), } - // Prepare the trigger socket. - let _ = std::fs::remove_file(&cfg.trigger_socket); - if let Some(parent) = cfg.trigger_socket.parent() { - std::fs::create_dir_all(parent).ok(); - } - let trigger = UnixListener::bind(&cfg.trigger_socket) + // Prepare the trigger socket. `bind_socket` publishes it 0600 and + // replaces any stale path atomically, so there is no window in which the + // trigger exists with umask-derived permissions. + let trigger = bind_socket(&cfg.trigger_socket) .with_context(|| format!("bind trigger socket {}", cfg.trigger_socket.display()))?; tracing::info!( @@ -149,7 +174,9 @@ fn main() -> Result<()> { "supervisor running" ); - // Trigger loop. One client at a time; commands are line-delimited. + // Trigger loop. One client at a time; commands are line-delimited. No + // failure below is allowed to escape: the supervisor outlives every + // client, so a bad connection logs and yields to the next one. for client in trigger.incoming() { let stream = match client { Ok(s) => s, @@ -158,26 +185,57 @@ fn main() -> Result<()> { continue; } }; - let mut writer = stream.try_clone()?; - let mut reader = BufReader::new(stream); - let mut line = String::new(); - if reader.read_line(&mut line).is_err() { + let Some(cmd) = read_trigger_command(&stream, &cfg) else { continue; - } - let cmd = line.trim(); - match handle_trigger(cmd, &cfg, &sup, ¤t_child) { - Ok(reply) => { - let _ = writeln!(writer, "{reply}"); - } - Err(e) => { - let _ = writeln!(writer, "err: {e}"); - } - } + }; + let reply = match handle_trigger(cmd.trim(), &cfg, &sup, ¤t_child) { + Ok(reply) => reply, + Err(e) => format!("err: {e}"), + }; + let mut writer = &stream; + let _ = writeln!(writer, "{reply}"); } Ok(()) } +/// Authenticate a trigger client and read its command line, or `None` if the +/// connection should be dropped. Every I/O step is time-bounded so one +/// misbehaving client cannot stall the single-threaded loop. +fn read_trigger_command(stream: &UnixStream, cfg: &Config) -> Option { + let uid = match peer_uid(stream) { + Ok(uid) => uid, + Err(e) => { + tracing::warn!(error = %e, "could not read trigger peer credentials; dropping"); + return None; + } + }; + if !peer_uid_is_allowed(uid, &cfg.allowed_uids) { + tracing::warn!(peer_uid = uid, "refusing trigger from unauthorized uid"); + let mut writer = stream; + let _ = writeln!(writer, "err: not permitted"); + return None; + } + if let Err(e) = stream + .set_read_timeout(Some(TRIGGER_IO_TIMEOUT)) + .and_then(|()| stream.set_write_timeout(Some(TRIGGER_IO_TIMEOUT))) + { + tracing::warn!(error = %e, "could not bound trigger client I/O; dropping"); + return None; + } + + let mut line = String::new(); + let mut reader = BufReader::new(stream.take(TRIGGER_MAX_LINE)); + match reader.read_line(&mut line) { + Ok(0) => None, + Ok(_) => Some(line), + Err(e) => { + tracing::warn!(error = %e, peer_uid = uid, "trigger command read failed"); + None + } + } +} + fn handle_trigger( cmd: &str, cfg: &Config, @@ -188,10 +246,10 @@ fn handle_trigger( let mut tokens = cmd.split_whitespace(); match tokens.next() { Some("handoff") => { - let binary = tokens - .next() - .map(PathBuf::from) - .unwrap_or_else(|| cfg.binary.clone()); + let binary = match tokens.next() { + Some(requested) => resolve_requested_binary(Path::new(requested), cfg)?, + None => cfg.binary.clone(), + }; let spec = SpawnSpec { binary, args: cfg.args.clone(), @@ -244,6 +302,33 @@ fn handle_trigger( } } +/// Map a client-supplied binary path onto the configured allowlist. +/// +/// The trigger socket is reachable by a local user, and `perform_handoff` +/// execs whatever path it is handed — so an unrestricted override is a +/// "run this file as the supervisor's user" primitive. Paths are compared +/// after canonicalization so `./foo`, `/srv/../srv/foo`, and a symlink to an +/// allowed target are all recognized as the same file. +fn resolve_requested_binary(requested: &Path, cfg: &Config) -> Result { + let canonical = requested + .canonicalize() + .with_context(|| format!("resolve requested binary {}", requested.display()))?; + let permitted = std::iter::once(&cfg.binary) + .chain(cfg.allowed_binaries.iter()) + .any(|allowed| { + allowed + .canonicalize() + .is_ok_and(|allowed| allowed == canonical) + }); + if !permitted { + anyhow::bail!( + "binary {} is not in the configured allowlist (`binary` + `allowed_binaries`)", + requested.display() + ); + } + Ok(canonical) +} + /// Poll-wait on a child up to `timeout`. Returns `Some(result)` if the /// child exited within the window, `None` if the timeout expired first. /// 50 ms poll interval — child exits are rare and the cost of one extra diff --git a/crates/handoff/src/error.rs b/crates/handoff/src/error.rs index 147b9dd..a836260 100644 --- a/crates/handoff/src/error.rs +++ b/crates/handoff/src/error.rs @@ -3,6 +3,7 @@ //! All fallible operations in the library return [`Result`]. The error enum //! is `Send` so it can flow across the consumer's runtime-bridging channels. +use std::path::PathBuf; use std::sync::mpsc::{RecvError, RecvTimeoutError, SendError}; #[derive(Debug, thiserror::Error)] @@ -51,6 +52,20 @@ pub enum Error { #[error("refused to break lock held by live process {holder_pid}")] StaleLockBreakRefused { holder_pid: i32 }, + #[error( + "unix socket path {} is {len} bytes, which exceeds the {max}-byte \ + sun_path limit on this platform", + .path.display() + )] + SocketPathTooLong { + path: PathBuf, + len: usize, + max: usize, + }, + + #[error("control socket peer uid {peer_uid} is not permitted")] + PeerNotPermitted { peer_uid: u32 }, + #[error("handoff already in progress")] HandoffInProgress, diff --git a/crates/handoff/src/frame.rs b/crates/handoff/src/frame.rs index 5c2fb6a..562e637 100644 --- a/crates/handoff/src/frame.rs +++ b/crates/handoff/src/frame.rs @@ -11,10 +11,12 @@ //! Frames are bounded by [`MAX_FRAME_BYTES`] to keep a malicious or buggy peer //! from triggering an unbounded allocation on the reader side. -use std::io::{Read, Write}; +use std::io::{ErrorKind, Read, Write}; +use std::os::unix::net::UnixStream; use crate::error::{Error, Result}; use crate::protocol::{Message, ProtoVersion}; +use crate::sock::send_all; /// Hard cap on a single frame's `frame_len`. 1 MiB is far larger than any /// legitimate handoff message; receipts above this are treated as malformed. @@ -25,8 +27,8 @@ const LEN_PREFIX: usize = 4; /// Size of the `proto_version` field that lives inside `frame_len`. const VERSION_FIELD: usize = 2; -/// Encode and write one `Message` to `w`. Flushes before returning. -pub fn write_message(w: &mut W, version: ProtoVersion, msg: &Message) -> Result<()> { +/// Encode one `Message` into a single contiguous frame buffer. +fn encode(version: ProtoVersion, msg: &Message) -> Result> { let payload = postcard::to_allocvec(msg)?; let inner_len = VERSION_FIELD .checked_add(payload.len()) @@ -34,14 +36,39 @@ pub fn write_message(w: &mut W, version: ProtoVersion, msg: &Message) if inner_len > MAX_FRAME_BYTES as usize { return Err(Error::FrameTooLarge(inner_len as u32)); } - let frame_len = inner_len as u32; - w.write_all(&frame_len.to_le_bytes())?; - w.write_all(&version.to_le_bytes())?; - w.write_all(&payload)?; + let mut out = Vec::with_capacity(LEN_PREFIX + inner_len); + out.extend_from_slice(&(inner_len as u32).to_le_bytes()); + out.extend_from_slice(&version.to_le_bytes()); + out.extend_from_slice(&payload); + Ok(out) +} + +/// Encode and write one `Message` to `w`. Flushes before returning. +/// +/// Prefer [`write_frame`] when the sink is the control socket: it cannot +/// raise `SIGPIPE` and emits the frame in one syscall. +pub fn write_message(w: &mut W, version: ProtoVersion, msg: &Message) -> Result<()> { + w.write_all(&encode(version, msg)?)?; w.flush()?; Ok(()) } +/// Write one `Message` to a control socket as a single `send(2)`. +/// +/// Two properties the generic [`write_message`] cannot offer: +/// +/// - **No `SIGPIPE`.** See [`crate::sock::send_all`] — a peer that died mid +/// handoff yields `EPIPE`, never a signal, whatever the embedding binary's +/// signal disposition happens to be. +/// - **One syscall per frame.** The three-`write_all` form could interleave +/// with a heartbeat thread's frame if the single-writer contract were ever +/// broken; a single `send` on a `SOCK_STREAM` Unix socket keeps the header +/// and payload contiguous in the buffer under any write ordering. +pub fn write_frame(stream: &UnixStream, version: ProtoVersion, msg: &Message) -> Result<()> { + send_all(stream, &encode(version, msg)?)?; + Ok(()) +} + /// Read one `Message` from `r`. Blocks until a complete frame is consumed or /// the stream returns an error / EOF. pub fn read_message(r: &mut R) -> Result<(ProtoVersion, Message)> { @@ -67,6 +94,104 @@ pub fn read_message(r: &mut R) -> Result<(ProtoVersion, Message)> { Ok((version, msg)) } +/// Incremental frame reader for sockets that carry a receive timeout. +/// +/// [`read_message`] is built on `read_exact`, which discards whatever it had +/// already consumed when the read fails. On a socket armed with `SO_RCVTIMEO` +/// that is a correctness bug, not just lost work: a frame that straddles the +/// timeout boundary leaves the stream mid-frame, and the next read +/// interprets payload bytes as a length prefix. Every subsequent frame is +/// garbage — reported as a malformed frame or, worse, a plausible-looking +/// message. +/// +/// The accumulator owns the partial bytes instead, so a timeout is a +/// *suspension*: the caller decides whether to keep waiting (peer is slow but +/// alive, and the wall-clock budget has room) or give up, and a resumed read +/// continues exactly where the previous one stopped. +#[derive(Debug, Default)] +pub struct FrameAccumulator { + buf: Vec, + /// Bytes needed for the frame in progress: [`LEN_PREFIX`] until the + /// length prefix has been parsed, `LEN_PREFIX + frame_len` after. + want: usize, +} + +impl FrameAccumulator { + pub fn new() -> Self { + Self { + buf: Vec::new(), + want: LEN_PREFIX, + } + } + + /// True if bytes of an incomplete frame are buffered. Callers use this to + /// distinguish "peer has gone silent" (nothing buffered — treat a timeout + /// as peer-dead) from "peer is mid-frame" (bytes buffered — the peer is + /// demonstrably alive, so keep reading until the wall-clock budget runs + /// out). + pub fn has_partial(&self) -> bool { + !self.buf.is_empty() + } + + /// Read toward the next complete message. Returns `Ok(None)` when the + /// read timed out (or would block) before the frame was complete; any + /// bytes consumed so far are retained for the next call. + pub fn poll_read(&mut self, r: &mut R) -> Result> { + if self.want == 0 { + self.want = LEN_PREFIX; + } + loop { + while self.buf.len() < self.want { + let start = self.buf.len(); + self.buf.resize(self.want, 0); + match r.read(&mut self.buf[start..]) { + Ok(0) => { + self.buf.truncate(start); + return Err(Error::Io(std::io::Error::from(ErrorKind::UnexpectedEof))); + } + Ok(n) => self.buf.truncate(start + n), + Err(e) if e.kind() == ErrorKind::Interrupted => { + self.buf.truncate(start); + } + Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => { + self.buf.truncate(start); + return Ok(None); + } + Err(e) => { + self.buf.truncate(start); + return Err(e.into()); + } + } + } + + if self.want == LEN_PREFIX { + let frame_len = + u32::from_le_bytes([self.buf[0], self.buf[1], self.buf[2], self.buf[3]]); + if frame_len < VERSION_FIELD as u32 { + self.reset(); + return Err(Error::FrameMalformed(frame_len)); + } + if frame_len > MAX_FRAME_BYTES { + self.reset(); + return Err(Error::FrameTooLarge(frame_len)); + } + self.want = LEN_PREFIX + frame_len as usize; + continue; + } + + let version = u16::from_le_bytes([self.buf[4], self.buf[5]]); + let decoded = postcard::from_bytes(&self.buf[LEN_PREFIX + VERSION_FIELD..]); + self.reset(); + return Ok(Some((version, decoded?))); + } + } + + fn reset(&mut self) { + self.buf.clear(); + self.want = LEN_PREFIX; + } +} + #[cfg(test)] mod tests { use std::io::Cursor; @@ -180,8 +305,129 @@ mod tests { assert!(read_message(&mut cursor).is_err()); } + /// A reader that hands out `chunks` in order and reports `TimedOut` once + /// each chunk is exhausted — the shape a `SO_RCVTIMEO`-armed socket + /// presents when the writer is slow or a frame is split across segments. + struct ChunkedTimeoutReader { + chunks: std::collections::VecDeque>, + pending_timeout: bool, + } + + impl ChunkedTimeoutReader { + fn new(chunks: impl IntoIterator>) -> Self { + Self { + chunks: chunks.into_iter().collect(), + pending_timeout: false, + } + } + } + + impl Read for ChunkedTimeoutReader { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + if self.pending_timeout || self.chunks.is_empty() { + self.pending_timeout = false; + return Err(std::io::Error::from(ErrorKind::TimedOut)); + } + let chunk = self + .chunks + .pop_front() + .expect("emptiness checked immediately above"); + let n = chunk.len().min(buf.len()); + buf[..n].copy_from_slice(&chunk[..n]); + if n < chunk.len() { + self.chunks.push_front(chunk[n..].to_vec()); + } else { + self.pending_timeout = true; + } + Ok(n) + } + } + + #[test] + fn accumulator_resumes_a_frame_split_across_timeouts() { + let msg = Message::SealComplete { + handoff_id: HandoffId::new(), + last_revision_per_shard: vec![1, 2, 3], + data_dir_fingerprint: [9u8; 32], + }; + let mut bytes = Vec::new(); + write_message(&mut bytes, PROTO_MAX, &msg).unwrap(); + + // Split mid-length-prefix and again mid-payload, with a receive + // timeout at each boundary: both points desynchronize a + // `read_exact`-based reader, which consumes the bytes it did get and + // then restarts parsing mid-frame on the next call. + let mut reader = ChunkedTimeoutReader::new([ + bytes[..2].to_vec(), + bytes[2..7].to_vec(), + bytes[7..].to_vec(), + ]); + + let mut acc = FrameAccumulator::new(); + assert!(acc.poll_read(&mut reader).unwrap().is_none()); + assert!(acc.has_partial(), "partial bytes must be retained"); + assert!(acc.poll_read(&mut reader).unwrap().is_none()); + assert!(acc.has_partial()); + let (ver, decoded) = acc + .poll_read(&mut reader) + .unwrap() + .expect("frame completes once the last chunk arrives"); + assert_eq!(ver, PROTO_MAX); + assert!(matches!(decoded, Message::SealComplete { .. })); + assert!(!acc.has_partial()); + + // Drained: a further poll is a plain timeout with nothing buffered, + // which is how callers recognize a silent peer as opposed to a slow + // one. + assert!(acc.poll_read(&mut reader).unwrap().is_none()); + assert!(!acc.has_partial()); + } + + #[test] + fn accumulator_decodes_back_to_back_frames_from_one_chunk() { + let mut bytes = Vec::new(); + write_message(&mut bytes, PROTO_MAX, &Message::Heartbeat { ts_ms: 1 }).unwrap(); + write_message(&mut bytes, PROTO_MAX, &Message::Heartbeat { ts_ms: 2 }).unwrap(); + let mut reader = Cursor::new(bytes); + let mut acc = FrameAccumulator::new(); + for expected in [1u64, 2] { + match acc.poll_read(&mut reader).unwrap() { + Some((_, Message::Heartbeat { ts_ms })) => assert_eq!(ts_ms, expected), + other => panic!("expected heartbeat {expected}, got {other:?}"), + } + } + } + + #[test] + fn accumulator_rejects_malformed_length_and_resets() { + let mut buf = (MAX_FRAME_BYTES + 1).to_le_bytes().to_vec(); + buf.extend_from_slice(&1u32.to_le_bytes()); + let mut reader = Cursor::new(buf); + let mut acc = FrameAccumulator::new(); + assert!(matches!( + acc.poll_read(&mut reader), + Err(Error::FrameTooLarge(_)) + )); + // Reset after the error: no stale partial keeps the caller from + // reusing the accumulator on a fresh connection. + assert!(!acc.has_partial()); + } + + #[test] + fn accumulator_reports_eof_mid_frame() { + let mut bytes = Vec::new(); + write_message(&mut bytes, PROTO_MAX, &Message::Heartbeat { ts_ms: 1 }).unwrap(); + bytes.truncate(bytes.len() - 1); + let mut reader = Cursor::new(bytes); + let mut acc = FrameAccumulator::new(); + match acc.poll_read(&mut reader) { + Err(Error::Io(e)) => assert_eq!(e.kind(), ErrorKind::UnexpectedEof), + other => panic!("expected UnexpectedEof, got {other:?}"), + } + } + proptest::proptest! { - /// Pure fuzz: `read_message` must never panic on any input bytes, + /// Pure fuzz: neither reader may panic on any input bytes, /// regardless of length, content, or alignment. The legal outcomes /// are `Ok(_)` (if the bytes happen to encode a valid frame) or /// `Err(_)` of any variant. @@ -189,8 +435,11 @@ mod tests { fn read_message_never_panics_on_arbitrary_bytes( bytes in proptest::collection::vec(proptest::num::u8::ANY, 0..2048), ) { - let mut cursor = Cursor::new(bytes); + let mut cursor = Cursor::new(bytes.clone()); let _ = read_message(&mut cursor); + let mut cursor = Cursor::new(bytes); + let mut acc = FrameAccumulator::new(); + let _ = acc.poll_read(&mut cursor); } /// Length prefix is honest: if we declare `frame_len = N` and feed diff --git a/crates/handoff/src/incumbent.rs b/crates/handoff/src/incumbent.rs index 7beefbb..41a63e5 100644 --- a/crates/handoff/src/incumbent.rs +++ b/crates/handoff/src/incumbent.rs @@ -22,13 +22,14 @@ use crate::crash::points; use crate::crash_here; use crate::drainable::Drainable; use crate::error::{Error, Result}; -use crate::frame::{read_message, write_message}; +use crate::frame::{FrameAccumulator, write_frame}; use crate::lock::DataDirLock; use crate::metrics::events; use crate::protocol::{ Capabilities, HandoffId, Message, PROTO_MAX, PROTO_MIN, ProtoVersion, Side, negotiate_version, short_name, }; +use crate::sock; use crate::util::now_unix_ms; /// How long the session-error recovery path will wait for the data-dir @@ -53,6 +54,16 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(2); /// otherwise pin the single-session serve loop indefinitely. const HELLO_READ_TIMEOUT: Duration = Duration::from_secs(5); +/// First backoff after an accept failure caused by resource exhaustion, and +/// the ceiling it doubles up to. Exhaustion (`EMFILE`, `ENFILE`, `ENOBUFS`, +/// `ENOMEM`) does not clear instantly and the pending connection stays in the +/// backlog, so retrying immediately would spin a core at full tilt while the +/// condition persists. Backing off costs a little latency on the (rare) +/// handoff connection and leaves the CPU available to whatever has to release +/// the descriptors. +const ACCEPT_BACKOFF_MIN: Duration = Duration::from_millis(5); +const ACCEPT_BACKOFF_MAX: Duration = Duration::from_secs(1); + pub struct Incumbent { listener: UnixListener, lock: Option, @@ -92,7 +103,6 @@ where }; let (stop_tx, stop_rx) = mpsc::channel::<()>(); let hb_thread = thread::spawn(move || { - let mut writer = writer; // `recv_timeout` returns Err on timeout — that's our "no stop // signal yet, send another heartbeat" trigger. `Ok(())` means // the main thread asked us to stop; any other Err means the @@ -101,7 +111,7 @@ where let msg = Message::Heartbeat { ts_ms: now_unix_ms(), }; - if write_message(&mut writer, chosen, &msg).is_err() { + if write_frame(&writer, chosen, &msg).is_err() { // Supervisor gone or socket broken — no point continuing. return; } @@ -133,21 +143,19 @@ where work() } -/// Bind the control socket, unlinking any prior path binding first. +/// Bind the control socket, taking over the path from any prior binding. /// Shared by `Incumbent::bind_cold_start` (no prior incumbent to displace) /// and `Incumbent::bind_after_ready` (called from a successor immediately /// after `Ready` so the prior incumbent is committed and exiting). The -/// preconditions are caller-enforced; this routine assumes the unlink is -/// safe at the call site. -fn bind_unlinking(socket_path: &Path, lock: DataDirLock) -> Result { - if let Some(parent) = socket_path.parent() { - std::fs::create_dir_all(parent)?; - } - // Remove any stale (cold start) or about-to-be-orphaned (after-ready) - // socket file. The caller has established that no live peer is - // serving on this path. - let _ = std::fs::remove_file(socket_path); - let listener = UnixListener::bind(socket_path)?; +/// preconditions are caller-enforced; this routine assumes displacing +/// whatever holds the path is safe at the call site. +fn bind_replacing(socket_path: &Path, lock: DataDirLock) -> Result { + // Bind onto a staging name and `rename(2)` it into place. This replaces + // any stale (cold start) or about-to-be-orphaned (after-ready) binding + // without the window in which the path does not exist, and publishes the + // socket only once its mode is `0600`. The caller has established that no + // live peer is serving on this path. + let listener = sock::bind_socket(socket_path)?; let data_dir = lock.data_dir().to_path_buf(); Ok(Incumbent { listener, @@ -175,6 +183,45 @@ fn acquire_with_short_retry(data_dir: &Path, timeout: Duration) -> Result AcceptFailure { + if e.kind() == ErrorKind::Interrupted { + return AcceptFailure::Retry; + } + match e.raw_os_error() { + Some(libc::ECONNABORTED) | Some(libc::EPROTO) | Some(libc::EPERM) => AcceptFailure::Retry, + // `EAGAIN` should not occur on our blocking listener, but if the + // consumer handed us a non-blocking one it must not become a spin: + // backing off makes the loop poll instead. + Some(libc::EMFILE) | Some(libc::ENFILE) | Some(libc::ENOBUFS) | Some(libc::ENOMEM) + | Some(libc::EAGAIN) => AcceptFailure::Exhausted, + _ => AcceptFailure::Fatal, + } +} + /// What happened to one supervisor session. enum SessionOutcome { /// Handoff committed — N is now the writer; this process should exit. @@ -215,7 +262,7 @@ impl Incumbent { /// [`Successor::announce_and_bind`], which orders `Ready` and bind in /// one call. pub fn bind_cold_start(socket_path: &Path, lock: DataDirLock) -> Result { - bind_unlinking(socket_path, lock) + bind_replacing(socket_path, lock) } /// Successor-side bind, called by [`crate::BegunSuccessor::announce_and_bind`] @@ -228,7 +275,7 @@ impl Incumbent { /// separate entry point exists so callers see a name that reflects the /// preconditions appropriate to their context. pub(crate) fn bind_after_ready(socket_path: &Path, lock: DataDirLock) -> Result { - bind_unlinking(socket_path, lock) + bind_replacing(socket_path, lock) } /// Set the implementation-defined build identifier announced in `Hello`. @@ -239,12 +286,37 @@ impl Incumbent { } pub fn serve(mut self, drainable: D) -> Result<()> { + let mut backoff = ACCEPT_BACKOFF_MIN; loop { let (stream, _addr) = match self.listener.accept() { - Ok(x) => x, - Err(e) if e.kind() == ErrorKind::Interrupted => continue, - Err(e) => return Err(e.into()), + Ok(x) => { + backoff = ACCEPT_BACKOFF_MIN; + x + } + Err(e) => match classify_accept_error(&e) { + AcceptFailure::Retry => continue, + AcceptFailure::Exhausted => { + // Descriptor/memory exhaustion: the daemon is still + // healthy and serving, and the condition is usually + // transient. Returning here would end the serve loop + // and leave a live process that can never be handed + // off again — the failure mode this backoff exists to + // avoid. + tracing::warn!( + error = %e, backoff_ms = backoff.as_millis() as u64, + "accept failed due to resource exhaustion; backing off" + ); + thread::sleep(backoff); + backoff = (backoff * 2).min(ACCEPT_BACKOFF_MAX); + continue; + } + AcceptFailure::Fatal => return Err(e.into()), + }, }; + if let Err(e) = self.prepare_session_stream(&stream) { + tracing::warn!(error = %e, "rejecting control connection"); + continue; + } match self.handle_session(stream, &drainable) { Ok(SessionOutcome::Committed) => { tracing::info!("handoff committed; incumbent exiting serve loop"); @@ -286,6 +358,23 @@ impl Incumbent { } } + /// Authenticate and configure a freshly accepted control connection. + /// + /// The socket's `0600` mode already keeps other users out on any sane + /// filesystem, but mode bits are advisory here in a way peer credentials + /// are not: an operator can loosen them, and some deployments put the + /// socket on a filesystem that ignores them. Since a control connection + /// can drain and seal the daemon, the uid the kernel latched at + /// `connect(2)` is checked directly. + fn prepare_session_stream(&self, stream: &UnixStream) -> Result<()> { + let uid = sock::peer_uid(stream)?; + if !sock::peer_uid_is_allowed(uid, &[]) { + return Err(Error::PeerNotPermitted { peer_uid: uid }); + } + sock::configure_control_stream(stream, sock::CONTROL_WRITE_TIMEOUT)?; + Ok(()) + } + fn handle_session( &mut self, mut stream: UnixStream, @@ -300,22 +389,24 @@ impl Incumbent { proto_max: PROTO_MAX, capabilities: Capabilities::default(), }; - write_message(&mut stream, PROTO_MAX, &our_hello)?; + write_frame(&stream, PROTO_MAX, &our_hello)?; // Receive HelloAck. Bound the read so a peer that connected and then // stalled without responding can't pin the serve loop. `serve()` // accepts one session at a time, so a single stuck peer would // otherwise block every legitimate handoff. + // + // The accumulator carries any partial frame past the timeout: a peer + // that sent half a `HelloAck` before the deadline expired is answered + // by the deadline, not by a desynchronized reader that would misread + // the remaining bytes as the next frame's length. + let mut acc = FrameAccumulator::new(); stream.set_read_timeout(Some(HELLO_READ_TIMEOUT))?; - let read_result = read_message(&mut stream); + let read_result = acc.poll_read(&mut stream); let _ = stream.set_read_timeout(None); let (_v, ack) = match read_result { - Ok(x) => x, - Err(Error::Io(e)) - if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => - { - return Err(Error::Timeout("HelloAck")); - } + Ok(Some(x)) => x, + Ok(None) => return Err(Error::Timeout("HelloAck")), Err(e) => return Err(e), }; let chosen = match ack { @@ -332,7 +423,7 @@ impl Incumbent { }; let mut state = SessionState::default(); - let outcome = self.run_session_loop(&mut stream, chosen, drainable, &mut state); + let outcome = self.run_session_loop(&mut stream, &mut acc, chosen, drainable, &mut state); // Drain-without-commit cleanup. The consumer stopped accepting when we // called `drain`; we need to tell them to start again before we leave @@ -354,13 +445,18 @@ impl Incumbent { fn run_session_loop( &mut self, stream: &mut UnixStream, + acc: &mut FrameAccumulator, chosen: u16, drainable: &D, state: &mut SessionState, ) -> Result { loop { - let (_v, msg) = match read_message(stream) { - Ok(x) => x, + // No read timeout is armed here, so `poll_read` blocks until a + // whole frame arrives; `Ok(None)` is unreachable but is treated + // as "keep waiting" rather than asserted away. + let (_v, msg) = match acc.poll_read(stream) { + Ok(Some(x)) => x, + Ok(None) => continue, Err(Error::Io(e)) if matches!( e.kind(), @@ -412,7 +508,7 @@ impl Incumbent { %handoff_id, open_conns_remaining = report.open_conns_remaining, "drain done" ); - write_message( + write_frame( stream, chosen, &Message::Drained { @@ -448,7 +544,7 @@ impl Incumbent { target: events::SEAL_COMPLETE, %handoff_id, "seal complete; flock released" ); - write_message( + write_frame( stream, chosen, &Message::SealComplete { @@ -463,7 +559,7 @@ impl Incumbent { tracing::error!( %handoff_id, error = %e, "seal failed; remaining as incumbent" ); - write_message( + write_frame( stream, chosen, &Message::SealFailed { @@ -542,7 +638,7 @@ impl Incumbent { state.active = None; } Message::Heartbeat { .. } => { - write_message( + write_frame( stream, chosen, &Message::Heartbeat { @@ -555,3 +651,71 @@ impl Incumbent { } } } + +#[cfg(test)] +mod tests { + use std::io::Error as IoError; + + use super::*; + + #[test] + fn transient_accept_errors_do_not_end_the_serve_loop() { + // The failure this guards against: a daemon that is alive and serving + // traffic but has left its accept loop, so no future handoff can ever + // reach it. + for errno in [libc::ECONNABORTED, libc::EPROTO, libc::EPERM, libc::EINTR] { + assert!( + matches!( + classify_accept_error(&IoError::from_raw_os_error(errno)), + AcceptFailure::Retry + ), + "errno {errno} should be retried immediately" + ); + } + for errno in [ + libc::EMFILE, + libc::ENFILE, + libc::ENOBUFS, + libc::ENOMEM, + libc::EAGAIN, + ] { + assert!( + matches!( + classify_accept_error(&IoError::from_raw_os_error(errno)), + AcceptFailure::Exhausted + ), + "errno {errno} should back off, not exit and not spin" + ); + } + for errno in [libc::EBADF, libc::EINVAL, libc::ENOTSOCK, libc::EFAULT] { + assert!( + matches!( + classify_accept_error(&IoError::from_raw_os_error(errno)), + AcceptFailure::Fatal + ), + "errno {errno} means the listener is unusable" + ); + } + } + + #[test] + fn cold_start_bind_is_private_and_replaces_a_stale_path() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().unwrap(); + let sock_path = dir.path().join("nested").join("ctl.sock"); + std::fs::create_dir_all(sock_path.parent().unwrap()).unwrap(); + // A stale socket file left by a crashed predecessor. + drop(UnixListener::bind(&sock_path).unwrap()); + + let lock = DataDirLock::acquire(dir.path()).unwrap(); + let incumbent = Incumbent::bind_cold_start(&sock_path, lock).unwrap(); + assert_eq!( + std::fs::metadata(&sock_path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + // The new binding is the one reachable at the path. + UnixStream::connect(&sock_path).expect("path serves the new listener"); + drop(incumbent); + } +} diff --git a/crates/handoff/src/lib.rs b/crates/handoff/src/lib.rs index fa4581d..6e0f6ca 100644 --- a/crates/handoff/src/lib.rs +++ b/crates/handoff/src/lib.rs @@ -3,10 +3,11 @@ //! See the crate-root `ARCHITECTURE.md` for the wire protocol, state machine, //! and correctness invariants. This module re-exports the public surface. -// Crate-wide safety gates. `unsafe_code` is denied by default; the four -// modules that legitimately need it (FD inheritance, env mutation at -// single-threaded startup, post-fork crash injection, and `FromRawFd` on -// kernel-handed descriptors) opt back in with `#[allow(unsafe_code)]` and +// Crate-wide safety gates. `unsafe_code` is denied by default; the modules +// that legitimately need it (FD inheritance, env mutation at single-threaded +// startup, post-fork crash injection, raw socket options and peer-credential +// lookups, and `FromRawFd` on kernel-handed descriptors) opt back in with +// `#[allow(unsafe_code)]` and // carry per-block `// SAFETY:` comments. `unused_must_use` is denied so a // dropped `Result` becomes a hard error rather than a silent regression. #![deny(unsafe_code)] @@ -22,6 +23,7 @@ pub mod lock; pub mod metrics; pub mod protocol; pub mod role; +pub mod sock; pub mod state; pub mod supervisor; mod util; @@ -36,4 +38,5 @@ pub use role::{ BegunSuccessor, HandshookSuccessor, HeartbeatGuard, InheritedListeners, Role, Successor, detect_role, }; +pub use sock::{bind_socket, peer_uid, peer_uid_is_allowed, validate_socket_path}; pub use supervisor::{HandoffOutcome, SpawnSpec, Supervisor}; diff --git a/crates/handoff/src/lock.rs b/crates/handoff/src/lock.rs index 8b3ae73..5af7fdf 100644 --- a/crates/handoff/src/lock.rs +++ b/crates/handoff/src/lock.rs @@ -9,6 +9,31 @@ //! pidfile remains as a hint. [`DataDirLock::acquire_or_break_stale`] uses the //! pidfile + `kill(pid, 0)` liveness check to safely break orphaned locks //! without risk of two-writers. +//! +//! # Filesystem requirements +//! +//! `flock(2)` is only an exclusion mechanism if the kernel that serves the +//! lock file sees every contender. That holds for local filesystems (ext4, +//! xfs, btrfs, zfs, apfs, ufs) and is the supported configuration. +//! +//! It does **not** hold everywhere a data directory can be pointed: +//! +//! - **NFS.** Linux emulates `flock` on NFSv3+ via POSIX record locks, which +//! are per-*process* rather than per-*open-file-description*: a lock can be +//! dropped by an unrelated `close()` in the same process, and the semantics +//! differ from the local case in ways this crate's invariants depend on. +//! Older or misconfigured mounts (`nolock`, no `rpc.statd`) degrade to a +//! purely local lock, which two hosts will both "acquire". +//! - **CIFS/SMB, 9p, FUSE without lock forwarding, overlayfs upper layers on +//! such backends.** Same failure mode: the lock is local to one client. +//! - **Two containers bind-mounting the same host directory** are fine (same +//! kernel), but two *hosts* sharing storage over a network filesystem are +//! not. +//! +//! The consequence of a lock that does not exclude is precisely the failure +//! this crate exists to prevent — two writers on one data directory — and it +//! cannot be detected from inside the lock holder. Put the data directory on +//! a local filesystem. use std::fs::{File, OpenOptions}; use std::io::Write; @@ -133,7 +158,15 @@ fn read_pidfile(path: &Path) -> Option { } fn write_pid_atomic(path: &Path, pid: u32) -> Result<()> { - let tmp = path.with_extension("pidfile.tmp"); + // Unique temp name per writer. A fixed `.tmp` is a shared + // mutable path: a second process writing its pidfile concurrently (the + // successor takes the flock while the exiting incumbent still holds a + // reference to the same data dir) would truncate the first writer's temp + // file mid-write, and either could then rename a half-written or + // wrong-pid file into place. The flock keeps that from being a + // correctness bug, but a pidfile naming the wrong process defeats + // `acquire_or_break_stale`, which is the whole point of having one. + let tmp = path.with_extension(format!("pidfile.{pid}.tmp")); { let mut f = OpenOptions::new() .write(true) @@ -143,7 +176,10 @@ fn write_pid_atomic(path: &Path, pid: u32) -> Result<()> { writeln!(f, "{pid}")?; f.sync_all()?; } - std::fs::rename(&tmp, path)?; + if let Err(e) = std::fs::rename(&tmp, path) { + let _ = std::fs::remove_file(&tmp); + return Err(e.into()); + } // fsync the parent directory so the rename's link-update is durable. // The pidfile is advisory (flock is authoritative), but a stale or // missing pidfile after crash recovery defeats `acquire_or_break_stale`'s @@ -159,11 +195,28 @@ fn write_pid_atomic(path: &Path, pid: u32) -> Result<()> { Ok(()) } +/// Whether `pid` names a live process. +/// +/// `kill(pid, 0)` reports three outcomes, and only one of them means "gone": +/// +/// - `Ok(())` — the process exists and we may signal it. +/// - `EPERM` — the process **exists** but belongs to another user. Reading +/// this as "dead" is the dangerous direction: it makes +/// [`DataDirLock::acquire_or_break_stale`] try to break a lock whose holder +/// is very much alive. (The flock still refuses, so this is a +/// misdiagnosis rather than a split brain — but it turns a clear +/// "held by a live process" into a confusing "stale lock could not be +/// broken", and it is a real configuration: a daemon restarted under a +/// different service account, or an operator recovering as a non-root user.) +/// - `ESRCH` — no such process. The only genuine death signal. fn is_pid_alive(pid: i32) -> bool { if pid <= 0 { return false; } - matches!(kill(Pid::from_raw(pid), None), Ok(())) + !matches!( + kill(Pid::from_raw(pid), None), + Err(nix::errno::Errno::ESRCH) + ) } #[cfg(test)] @@ -255,4 +308,35 @@ mod tests { // a genuinely-held flock, not a transient one. assert!(_other_flock.as_raw_fd() >= 0); } + + #[test] + fn a_process_we_may_not_signal_still_counts_as_alive() { + // PID 1 exists and (when we are not root) `kill(1, 0)` fails with + // EPERM. Reading that as "dead" would make the stale-break path try + // to reclaim a lock from a live holder. + assert!(is_pid_alive(1), "pid 1 is alive regardless of our uid"); + // ESRCH is the only genuine death signal. + assert!(!is_pid_alive(i32::MAX)); + assert!(!is_pid_alive(0)); + assert!(!is_pid_alive(-1)); + } + + #[test] + fn pidfile_temp_path_is_writer_specific() { + let dir = tempfile::tempdir().unwrap(); + let pid_path = dir.path().join(PID_FILE); + write_pid_atomic(&pid_path, 4242).unwrap(); + assert_eq!(std::fs::read_to_string(&pid_path).unwrap().trim(), "4242"); + // No temp file survives a successful write, and the name it used was + // not the shared `.tmp` another writer would also pick. + let leftovers: Vec<_> = std::fs::read_dir(dir.path()) + .unwrap() + .map(|e| e.unwrap().file_name()) + .filter(|n| n.to_string_lossy().ends_with(".tmp")) + .collect(); + assert!( + leftovers.is_empty(), + "temp files left behind: {leftovers:?}" + ); + } } diff --git a/crates/handoff/src/role.rs b/crates/handoff/src/role.rs index 2c8d7b5..c9821c4 100644 --- a/crates/handoff/src/role.rs +++ b/crates/handoff/src/role.rs @@ -7,9 +7,41 @@ //! - `HANDOFF_SOCK_FD=` — open Unix socket to supervisor //! - `LISTEN_FDS=` — count of inherited listener FDs (starting at FD 3) //! - `LISTEN_FDNAMES=resp:http:…` — colon-separated logical names in FD order +//! - `LISTEN_PID=` — optional; when present it must equal our pid //! //! [`detect_role`] reads these and consumes them so an accidental double-detect //! gives [`Role::ColdStart`] (which is what fresh re-execs should do). +//! +//! # Why the descriptors are validated before adoption +//! +//! `LISTEN_FDS`/`LISTEN_FDNAMES`/`LISTEN_PID` are systemd's variables, not +//! ours — we speak that convention so a socket-activated unit can cold-start a +//! handoff-aware daemon unchanged. The cost is that these names can arrive +//! from somewhere other than a handoff supervisor, and `FromRawFd` validates +//! nothing: a stale `LISTEN_FDS=3` inherited through an unrelated `execve` +//! would have us wrap FDs 3–5 — possibly a log file, a database connection, or +//! nothing at all — in `TcpListener`s and close them when they drop. +//! +//! Two defenses, in the order the kernel lets us apply them: +//! +//! 1. **`LISTEN_PID`.** systemd always sets it to the pid it is activating. +//! If it is present and names a different process, the whole block belongs +//! to an ancestor and is ignored. (The supervisor in this crate cannot set +//! it — the value is only knowable after `fork`, and `Command`'s +//! environment is materialized before `pre_exec` runs — so its absence is +//! not by itself suspicious. Successor identity is instead verified on the +//! wire by the `Hello` pid check.) +//! 2. **Per-descriptor validation.** Every slot must be an open socket +//! (`fstat` reports `S_IFSOCK`), and [`InheritedListeners::take`] / +//! [`InheritedListeners::take_unix`] additionally require the listening +//! state and the right address family before handing back a typed listener. +//! +//! Adopted descriptors are also normalized: `FD_CLOEXEC` is re-armed (the +//! parent's `dup2` cleared it so the FD could survive `execve`, but leaving it +//! clear leaks listeners and the control socket into every subprocess the +//! daemon later spawns) and `O_NONBLOCK` is cleared (it lives on the open file +//! description, so a parent that gave its listener to an async runtime would +//! otherwise hand us a listener whose first `accept()` returns `EAGAIN`). // Env mutation (`env::remove_var`, `env::set_var`) is `unsafe` in Rust 2024 // because it races with concurrent env reads in other threads; this module @@ -23,17 +55,18 @@ use std::env; use std::marker::PhantomData; use std::net::TcpListener; use std::os::fd::{FromRawFd, RawFd}; -use std::os::unix::net::UnixStream; +use std::os::unix::net::{UnixListener, UnixStream}; use std::sync::mpsc; use std::thread; use std::time::Duration; use crate::drainable::ReadinessSnapshot; use crate::error::{Error, Result}; -use crate::frame::{read_message, write_message}; +use crate::frame::{read_message, write_frame}; use crate::protocol::{ Capabilities, HandoffId, Message, PROTO_MAX, PROTO_MIN, ProtoVersion, Side, short_name, }; +use crate::sock; use crate::util::now_unix_ms; /// Cadence matched to the incumbent's heartbeat thread; with the @@ -45,6 +78,9 @@ pub const ENV_HANDOFF_ROLE: &str = "HANDOFF_ROLE"; pub const ENV_HANDOFF_SOCK_FD: &str = "HANDOFF_SOCK_FD"; pub const ENV_LISTEN_FDS: &str = "LISTEN_FDS"; pub const ENV_LISTEN_FDNAMES: &str = "LISTEN_FDNAMES"; +/// systemd sets this to the pid of the process it is activating. Honored when +/// present; see the module docs for why we never set it ourselves. +pub const ENV_LISTEN_PID: &str = "LISTEN_PID"; /// Inherited listener FDs start here, matching the systemd convention. pub const SD_LISTEN_FDS_START: RawFd = 3; @@ -68,14 +104,73 @@ pub struct InheritedListeners { } impl InheritedListeners { - /// Consume the inherited listener for `name`. Returns `None` if no such - /// listener was passed (or it was already taken). + /// Consume the inherited TCP listener for `name`. + /// + /// Returns `None` if no such listener was passed, if it was already taken, + /// or if the descriptor is not a listening `AF_INET`/`AF_INET6` socket — + /// the last case is logged at WARN and leaves the descriptor untouched + /// rather than wrapping (and eventually closing) a descriptor that belongs + /// to something else. Use [`Self::take_unix`] for `AF_UNIX` listeners. pub fn take(&mut self, name: &str) -> Option { - let fd = self.listeners.remove(name)?; - // SAFETY: kernel inherited the FD to us via fork+exec; we own it. + let fd = self.take_validated( + name, + &[ + libc::AF_INET as libc::sa_family_t, + libc::AF_INET6 as libc::sa_family_t, + ], + "TCP", + )?; + // SAFETY: validated just above as a live, listening AF_INET/AF_INET6 + // socket that the kernel handed us via fork+exec; nothing else in this + // process owns it, and the map entry is now gone. Some(unsafe { TcpListener::from_raw_fd(fd) }) } + /// Consume the inherited Unix-domain listener for `name`. + /// + /// The same inheritance convention covers `AF_UNIX` listeners — a daemon + /// serving on a Unix socket has exactly the same reason to keep its + /// binding across a handoff as one serving TCP, and rebinding the path + /// would drop connections queued in the backlog. + pub fn take_unix(&mut self, name: &str) -> Option { + let fd = self.take_validated(name, &[libc::AF_UNIX as libc::sa_family_t], "Unix")?; + // SAFETY: validated as a live, listening AF_UNIX socket owned by us. + Some(unsafe { UnixListener::from_raw_fd(fd) }) + } + + /// Shared validation for the typed `take*` accessors. Only removes the + /// entry when the descriptor really is a listening socket of one of + /// `families`, so a mismatched call can't silently discard a listener the + /// consumer will ask for again under the right accessor. + fn take_validated( + &mut self, + name: &str, + families: &[libc::sa_family_t], + kind: &str, + ) -> Option { + let fd = *self.listeners.get(name)?; + if !sock::fd_is_listening(fd) { + tracing::warn!( + name, + fd, + "inherited descriptor is not a listening socket; refusing to adopt it" + ); + return None; + } + match sock::socket_family(fd) { + Some(f) if families.contains(&f) => {} + other => { + tracing::warn!( + name, fd, family = ?other, + "inherited listener is not a {kind} socket; refusing to adopt it" + ); + return None; + } + } + self.listeners.remove(name); + Some(fd) + } + /// Names of all listeners that haven't yet been taken. pub fn names(&self) -> Vec { self.listeners.keys().cloned().collect() @@ -134,6 +229,11 @@ pub struct BegunSuccessor { /// listeners. Env vars are removed so re-entry yields a clean state. pub fn detect_role() -> Result { let inherited = read_inherited_listeners(); + // SAFETY: same single-threaded-startup invariant as the other env + // mutations in this function. + unsafe { + env::remove_var(ENV_LISTEN_PID); + } // SAFETY: `env::remove_var` races with concurrent env reads on other // threads (`std::env::set_var` / `getenv` from libc). `detect_role` is // contracted to run during single-threaded startup before the primitive @@ -164,12 +264,39 @@ pub fn detect_role() -> Result { env::remove_var(ENV_HANDOFF_SOCK_FD); } - // SAFETY: the supervisor handed us this FD via `fork+exec`. It's open and - // owned by us from here on. + if !sock::fd_is_socket(sock_fd) { + return Err(Error::BadEnv { + var: ENV_HANDOFF_SOCK_FD, + value: format!("fd {sock_fd} is not an open socket"), + }); + } + // The control socket needs the same normalization as the listeners: it + // must not leak into subprocesses (a leaked copy holds the supervisor's + // EOF open, hiding our death from it) and the protocol code assumes + // blocking reads. + normalize_inherited_fd(sock_fd, "control socket"); + + // SAFETY: the supervisor handed us this FD via `fork+exec` and we have + // just confirmed it is an open socket. It's owned by us from here on. let control = unsafe { UnixStream::from_raw_fd(sock_fd) }; + if let Err(e) = sock::configure_control_stream(&control, sock::CONTROL_WRITE_TIMEOUT) { + tracing::warn!(error = %e, "could not configure successor control socket"); + } Ok(Role::Successor(Successor { control, inherited })) } +/// Re-arm `FD_CLOEXEC` and clear `O_NONBLOCK` on an adopted descriptor. +/// Best-effort: a failure is logged and the descriptor is still usable, just +/// with the inherited flag state. +fn normalize_inherited_fd(fd: RawFd, what: &str) { + if let Err(e) = sock::set_cloexec(fd) { + tracing::warn!(fd, error = %e, "could not re-arm FD_CLOEXEC on inherited {what}"); + } + if let Err(e) = sock::clear_nonblocking(fd) { + tracing::warn!(fd, error = %e, "could not clear O_NONBLOCK on inherited {what}"); + } +} + fn read_inherited_listeners() -> InheritedListeners { let count: usize = env::var(ENV_LISTEN_FDS) .ok() @@ -178,6 +305,22 @@ fn read_inherited_listeners() -> InheritedListeners { if count == 0 { return InheritedListeners::default(); } + // systemd's activation contract: the FD block belongs to the pid named in + // LISTEN_PID. If it names someone else, we inherited the variables through + // an intervening exec and the descriptors are not ours to touch. + if let Ok(raw) = env::var(ENV_LISTEN_PID) { + let ours = std::process::id(); + match raw.trim().parse::() { + Ok(pid) if pid == ours => {} + other => { + tracing::warn!( + listen_pid = %raw, our_pid = ours, parsed = ?other.ok(), + "LISTEN_PID does not name this process; ignoring inherited listeners" + ); + return InheritedListeners::default(); + } + } + } let names: Vec = env::var(ENV_LISTEN_FDNAMES) .ok() .map(|s| s.split(':').map(|s| s.to_string()).collect()) @@ -186,6 +329,19 @@ fn read_inherited_listeners() -> InheritedListeners { for i in 0..count { let fd = SD_LISTEN_FDS_START + i as RawFd; let name = names.get(i).cloned().unwrap_or_else(|| i.to_string()); + if !sock::fd_is_socket(fd) { + // Either the count over-reports what was passed, or the variables + // reached us from an unrelated ancestor. Skipping is the only safe + // response: adopting the slot would hand a `TcpListener` a + // descriptor it does not own and close it on drop. + tracing::warn!( + name, + fd, + "LISTEN_FDS names a descriptor that is not an open socket; skipping" + ); + continue; + } + normalize_inherited_fd(fd, "listener"); map.insert(name, fd); } InheritedListeners { listeners: map } @@ -205,7 +361,7 @@ impl Successor { proto_max: PROTO_MAX, capabilities: Capabilities::default(), }; - write_message(&mut self.control, PROTO_MAX, &hello)?; + write_frame(&self.control, PROTO_MAX, &hello)?; let (_ver, ack) = read_message(&mut self.control)?; match ack { Message::HelloAck { @@ -281,6 +437,12 @@ impl BegunSuccessor { self.inherited.take(name) } + /// Consume the inherited Unix-domain listener for `name`. See + /// [`InheritedListeners::take_unix`]. + pub fn take_unix_listener(&mut self, name: &str) -> Option { + self.inherited.take_unix(name) + } + /// Names of inherited listeners that haven't yet been taken. pub fn listener_names(&self) -> Vec { self.inherited.names() @@ -308,14 +470,14 @@ impl BegunSuccessor { /// correctly by construction. Use this lower-level entry point only /// when you genuinely need to delay binding (e.g. for additional /// post-`Ready` setup that does not require the control socket). - pub fn announce_ready(mut self, snapshot: ReadinessSnapshot) -> Result<()> { + pub fn announce_ready(self, snapshot: ReadinessSnapshot) -> Result<()> { let ready = Message::Ready { handoff_id: self.handoff_id, listening_on: snapshot.listening_on, healthz_ok: snapshot.healthz_ok, advertised_revision_per_shard: snapshot.advertised_revision_per_shard, }; - write_message(&mut self.control, self.proto_version, &ready)?; + write_frame(&self.control, self.proto_version, &ready)?; Ok(()) } @@ -393,7 +555,6 @@ impl<'a> HeartbeatGuard<'a> { }; let (stop_tx, stop_rx) = mpsc::channel::<()>(); let thread = thread::spawn(move || { - let mut writer = writer; // `recv_timeout` returns Err on timeout — "no stop yet, send // another heartbeat". `Ok(())` or any other Err (sender // dropped) is the stop signal. @@ -401,7 +562,7 @@ impl<'a> HeartbeatGuard<'a> { let msg = Message::Heartbeat { ts_ms: now_unix_ms(), }; - if write_message(&mut writer, chosen, &msg).is_err() { + if write_frame(&writer, chosen, &msg).is_err() { // Supervisor gone or socket broken — no point continuing. return; } @@ -428,6 +589,8 @@ impl Drop for HeartbeatGuard<'_> { #[cfg(test)] mod tests { + use std::os::fd::AsRawFd; + use super::*; // Env mutation is process-global; run env-touching tests sequentially in @@ -455,5 +618,92 @@ mod tests { unsafe { env::remove_var(ENV_HANDOFF_ROLE); } + + // LISTEN_PID naming another process: the whole block belongs to an + // ancestor and must be ignored, however plausible LISTEN_FDS looks. + // SAFETY: same single-threaded-test invariant as above. + unsafe { + env::set_var(ENV_LISTEN_FDS, "3"); + env::set_var(ENV_LISTEN_FDNAMES, "http:grpc:admin"); + env::set_var(ENV_LISTEN_PID, (std::process::id() + 1).to_string()); + } + match detect_role().unwrap() { + Role::ColdStart { inherited } => assert!( + inherited.is_empty(), + "listeners adopted despite a foreign LISTEN_PID: {:?}", + inherited.names() + ), + _ => panic!("expected ColdStart"), + } + + // Matching LISTEN_PID: the block is ours, but every slot still has to + // prove it is an open socket before it lands in the map. Under a test + // harness those low FDs are the harness's own files and pipes, so a + // count that over-reports must never produce an adoptable entry. + // SAFETY: same single-threaded-test invariant as above. + unsafe { + env::set_var(ENV_LISTEN_FDS, "3"); + env::remove_var(ENV_LISTEN_FDNAMES); + env::set_var(ENV_LISTEN_PID, std::process::id().to_string()); + } + match detect_role().unwrap() { + Role::ColdStart { inherited } => { + for (name, fd) in &inherited.listeners { + assert!( + crate::sock::fd_is_socket(*fd), + "adopted non-socket fd {fd} as listener {name}" + ); + } + } + _ => panic!("expected ColdStart"), + } + + // SAFETY: same single-threaded-test invariant as above. + unsafe { + env::remove_var(ENV_LISTEN_FDS); + env::remove_var(ENV_LISTEN_PID); + } + } + + #[test] + fn take_and_take_unix_are_family_checked() { + let dir = tempfile::tempdir().unwrap(); + let unix = crate::sock::bind_socket(&dir.path().join("s.sock")).unwrap(); + let tcp = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut inherited = InheritedListeners { + listeners: HashMap::from([ + ("unix".to_string(), unix.as_raw_fd()), + ("tcp".to_string(), tcp.as_raw_fd()), + ]), + }; + + // Wrong accessor for the family: refuse, and keep the entry so the + // right accessor still works. + assert!(inherited.take("unix").is_none()); + assert!(inherited.take_unix("tcp").is_none()); + + let adopted_tcp = inherited.take("tcp").expect("tcp listener adoptable"); + let adopted_unix = inherited + .take_unix("unix") + .expect("unix listener adoptable"); + assert!(inherited.is_empty()); + // The adopted wrappers own the FDs now; forget the originals so the + // test doesn't double-close them on drop. + std::mem::forget(unix); + std::mem::forget(tcp); + drop((adopted_tcp, adopted_unix)); + } + + #[test] + fn take_refuses_a_connected_socket() { + // A connected (non-listening) socket in a listener slot means the + // count or the FD order is wrong; adopting it would produce a + // `TcpListener` whose `accept()` fails forever. + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let client = std::net::TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let mut inherited = InheritedListeners { + listeners: HashMap::from([("http".to_string(), client.as_raw_fd())]), + }; + assert!(inherited.take("http").is_none()); } } diff --git a/crates/handoff/src/sock.rs b/crates/handoff/src/sock.rs new file mode 100644 index 0000000..cc1cdc3 --- /dev/null +++ b/crates/handoff/src/sock.rs @@ -0,0 +1,550 @@ +//! Unix-domain socket helpers shared by the incumbent, the supervisor, and +//! the reference binary. +//! +//! Four concerns live here, all of them platform mechanics that the protocol +//! code above should not have to restate: +//! +//! - **`sun_path` validation.** `bind(2)` truncates or rejects paths longer +//! than `sockaddr_un.sun_path` (108 bytes on Linux, 104 on macOS/BSD). +//! Catching that up front turns a confusing `EINVAL`/silent-truncation into +//! [`Error::SocketPathTooLong`]. +//! - **Atomic, private bind.** [`bind_socket`] binds a temporary name in the +//! target directory, tightens the mode to `0600`, then `rename(2)`s it over +//! the final path. There is never a window in which the published path is +//! either missing (unlink-then-bind) or world-writable (bind-then-chmod). +//! - **Peer authentication.** [`peer_uid`] reads the connecting process's uid +//! from the kernel (`SO_PEERCRED` on Linux, `getpeereid(3)` elsewhere), so a +//! control socket can refuse commands from other local users even if the +//! filesystem permissions were loosened by an operator. +//! - **SIGPIPE-safe writes.** [`send_all`] uses `MSG_NOSIGNAL` where the +//! platform has it and `SO_NOSIGPIPE` where it doesn't, so a write to a +//! control socket whose peer just died surfaces as `EPIPE` instead of a +//! process-killing signal. + +// Every routine here is a thin wrapper over a libc socket call; the raw FFI +// is the point of the module. Each block carries its own SAFETY note. +#![allow(unsafe_code)] + +use std::ffi::c_void; +use std::io; +use std::mem::size_of; +use std::os::fd::{AsRawFd, RawFd}; +use std::os::unix::fs::PermissionsExt; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use crate::error::{Error, Result}; + +/// Mode applied to a bound control socket: owner read/write only. Peer +/// authentication ([`peer_uid`]) is the real access control; this keeps the +/// filesystem from being the weak link when the process runs under a lax +/// umask (the default `bind(2)` mode is `0777 & ~umask`, i.e. world-writable +/// under `umask 0`). +const SOCKET_MODE: u32 = 0o600; + +/// Bound on any single write to a control socket. Matches the supervisor's +/// per-recv liveness timeout: a peer that has not drained our frame within +/// the window it is allowed to go silent is unresponsive by the same +/// definition, and the handoff should fail rather than park a thread in an +/// unbounded `write(2)`. +pub const CONTROL_WRITE_TIMEOUT: Duration = Duration::from_secs(10); + +/// Usable bytes in `sockaddr_un.sun_path`, minus the NUL terminator. +pub fn max_socket_path_len() -> usize { + // SAFETY: reading the size of a field on a zeroed POD struct; no + // dereference of uninitialized memory beyond `size_of_val`. + let addr: libc::sockaddr_un = unsafe { std::mem::zeroed() }; + std::mem::size_of_val(&addr.sun_path) - 1 +} + +/// Reject a socket path the kernel could not represent in `sockaddr_un`. +/// +/// Callers should run this before any `bind`/`connect` so an over-long path +/// fails with a diagnosable error naming the limit, rather than an `EINVAL` +/// (Linux) or a silently truncated binding (some BSDs). +pub fn validate_socket_path(path: &Path) -> Result<()> { + let len = path.as_os_str().len(); + let max = max_socket_path_len(); + if len > max { + return Err(Error::SocketPathTooLong { + path: path.to_path_buf(), + len, + max, + }); + } + Ok(()) +} + +/// Bind a Unix listener at `path`, atomically and with mode `0600`. +/// +/// The bind happens on a temporary sibling name, which is chmod'ed and then +/// `rename(2)`d over `path`. Two properties follow: +/// +/// - **Atomic replacement.** A client connecting during a rebind sees either +/// the old binding or the new one, never `ENOENT`. The unlink-then-bind +/// alternative has a window where the path does not exist, and a second +/// binder racing in that window silently steals the name. +/// - **No permissive window.** The socket is never reachable at its published +/// path while its mode is still whatever `0777 & ~umask` produced. +/// +/// Any existing binding at `path` is replaced. Callers are responsible for +/// establishing that replacing it is safe (see [`crate::Incumbent`]). +pub fn bind_socket(path: &Path) -> Result { + validate_socket_path(path)?; + if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) { + std::fs::create_dir_all(parent)?; + } + let staging = staging_path(path)?; + validate_socket_path(&staging)?; + + // A leftover staging file (previous process killed mid-bind) is ours to + // remove: the name embeds our pid and a per-call counter. + let _ = std::fs::remove_file(&staging); + let listener = UnixListener::bind(&staging)?; + let bind_result = (|| -> io::Result<()> { + std::fs::set_permissions(&staging, std::fs::Permissions::from_mode(SOCKET_MODE))?; + std::fs::rename(&staging, path) + })(); + if let Err(e) = bind_result { + drop(listener); + let _ = std::fs::remove_file(&staging); + return Err(e.into()); + } + set_cloexec(listener.as_raw_fd())?; + Ok(listener) +} + +/// Sibling path used as the staging name for [`bind_socket`]. The file name +/// is deliberately short (pid + counter, not the full target name) so the +/// staging path stays inside `sun_path` even when the target is close to the +/// limit. +fn staging_path(path: &Path) -> Result { + use std::sync::atomic::{AtomicU32, Ordering}; + static SEQ: AtomicU32 = AtomicU32::new(0); + let seq = SEQ.fetch_add(1, Ordering::Relaxed); + let parent = path.parent().unwrap_or(Path::new(".")); + let parent = if parent.as_os_str().is_empty() { + Path::new(".") + } else { + parent + }; + Ok(parent.join(format!(".ho-{}-{seq}.tmp", std::process::id()))) +} + +/// The uid of the process on the other end of `stream`. +/// +/// Reported by the kernel at `connect(2)` time, so it cannot be forged by the +/// peer and does not race with the peer exiting (the credentials are latched +/// with the connection, not looked up per call). +#[cfg(any(target_os = "linux", target_os = "android"))] +pub fn peer_uid(stream: &UnixStream) -> io::Result { + let mut cred: libc::ucred = unsafe { std::mem::zeroed() }; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: `cred` is a live, correctly-sized `ucred`; `len` matches it. + let rc = unsafe { + libc::getsockopt( + stream.as_raw_fd(), + libc::SOL_SOCKET, + libc::SO_PEERCRED, + &mut cred as *mut libc::ucred as *mut c_void, + &mut len, + ) + }; + if rc == -1 { + return Err(io::Error::last_os_error()); + } + Ok(cred.uid) +} + +/// The uid of the process on the other end of `stream`. macOS and the BSDs +/// expose the latched credentials through `getpeereid(3)` rather than a +/// `SO_PEERCRED` socket option. +#[cfg(not(any(target_os = "linux", target_os = "android")))] +pub fn peer_uid(stream: &UnixStream) -> io::Result { + let mut uid: libc::uid_t = 0; + let mut gid: libc::gid_t = 0; + // SAFETY: both out-params are live locals of the expected types. + let rc = unsafe { libc::getpeereid(stream.as_raw_fd(), &mut uid, &mut gid) }; + if rc == -1 { + return Err(io::Error::last_os_error()); + } + Ok(uid) +} + +/// True if `uid` may drive the control protocol on this process's socket. +/// +/// Our own effective uid is allowed (the normal case: supervisor and +/// primitive run as the same service account) and so is root, which can +/// `ptrace`/`kill` us anyway — refusing it would buy nothing and would break +/// legitimate operator tooling. Everything else is rejected: a drain+seal is +/// a privileged operation on the daemon's data directory. +pub fn peer_uid_is_allowed(uid: u32, extra_allowed: &[u32]) -> bool { + // SAFETY: `geteuid` cannot fail and touches no memory. + let euid = unsafe { libc::geteuid() }; + uid == euid || uid == 0 || extra_allowed.contains(&uid) +} + +/// Set `FD_CLOEXEC` on `fd`, preserving other descriptor flags. +/// +/// Inherited descriptors arrive CLOEXEC-cleared (that is how they survived +/// `execve`); re-arming the flag keeps a listener or control socket from +/// leaking into every subprocess the daemon later spawns. A leaked control +/// socket is not merely untidy: it holds the peer's EOF open, so the +/// supervisor cannot detect that the primitive died. +pub fn set_cloexec(fd: RawFd) -> io::Result<()> { + // SAFETY: `fd` is a live descriptor owned by the caller; F_GETFD/F_SETFD + // only read and write that descriptor's flag word. + let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) }; + if flags == -1 { + return Err(io::Error::last_os_error()); + } + if flags & libc::FD_CLOEXEC != 0 { + return Ok(()); + } + // SAFETY: as above. + if unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// Clear `O_NONBLOCK` on `fd`, preserving other status flags. +/// +/// `O_NONBLOCK` lives on the *open file description*, not the descriptor, so +/// it survives `dup2` and `execve`. A parent that handed its listener to an +/// async runtime therefore passes a non-blocking listener to the child, whose +/// first `accept()` returns `EAGAIN` — a spurious "no connection" that a +/// blocking-API consumer has no reason to expect. +pub fn clear_nonblocking(fd: RawFd) -> io::Result<()> { + // SAFETY: `fd` is a live descriptor; F_GETFL/F_SETFL only touch its + // status flags. + let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + if flags == -1 { + return Err(io::Error::last_os_error()); + } + if flags & libc::O_NONBLOCK == 0 { + return Ok(()); + } + // SAFETY: as above. + if unsafe { libc::fcntl(fd, libc::F_SETFL, flags & !libc::O_NONBLOCK) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// True if `fd` is open and refers to a socket. Used before adopting an +/// inherited descriptor: `FromRawFd` performs no validation, so a bogus +/// `LISTEN_FDS` count would otherwise wrap an unrelated descriptor (or a +/// closed one) in a `TcpListener` and later close it out from under its real +/// owner. +pub fn fd_is_socket(fd: RawFd) -> bool { + let mut st: libc::stat = unsafe { std::mem::zeroed() }; + // SAFETY: `st` is a live, correctly-sized `stat`. `fstat` on a closed or + // invalid fd returns EBADF rather than misbehaving. + let rc = unsafe { libc::fstat(fd, &mut st) }; + rc == 0 && (st.st_mode & libc::S_IFMT) == libc::S_IFSOCK +} + +/// Address family of the socket behind `fd` (`AF_INET`, `AF_UNIX`, …), or +/// `None` if `fd` is not a bound socket. Read via `getsockname(2)` because +/// `SO_DOMAIN` is Linux-only. +pub fn socket_family(fd: RawFd) -> Option { + let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: `storage` is a live, correctly-sized `sockaddr_storage` and + // `len` describes it accurately. + let rc = unsafe { + libc::getsockname( + fd, + &mut storage as *mut libc::sockaddr_storage as *mut libc::sockaddr, + &mut len, + ) + }; + if rc == -1 { + return None; + } + Some(storage.ss_family) +} + +/// True if `fd` is a socket in the listening state. +/// +/// `SO_ACCEPTCONN` answers this directly on Linux. It is not dependable +/// elsewhere — XNU's `sogetopt` does not report it for every socket kind — +/// so on other Unices a negative answer falls back to the shape a listener +/// has: a stream socket with no peer. `getpeername(2)` returns `ENOTCONN` +/// for a listener and succeeds for a connected socket, which is the +/// distinction the callers actually need (a bound-but-not-listening socket +/// is indistinguishable this way, and merely surfaces later as an `accept` +/// error rather than as a silently adopted wrong descriptor). +#[cfg(any(target_os = "linux", target_os = "android"))] +pub fn fd_is_listening(fd: RawFd) -> bool { + accept_conn_opt(fd).unwrap_or(false) +} + +#[cfg(not(any(target_os = "linux", target_os = "android")))] +pub fn fd_is_listening(fd: RawFd) -> bool { + accept_conn_opt(fd).unwrap_or(false) + || (socket_type(fd) == Some(libc::SOCK_STREAM) && !socket_is_connected(fd)) +} + +/// `SO_ACCEPTCONN`, or `None` where the platform does not implement it. +fn accept_conn_opt(fd: RawFd) -> Option { + let mut val: libc::c_int = 0; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: `val`/`len` are live locals of the sizes the option expects. + let rc = unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_ACCEPTCONN, + &mut val as *mut libc::c_int as *mut c_void, + &mut len, + ) + }; + if rc == 0 { + return Some(val != 0); + } + // EBADF/ENOTSOCK are real answers: not a listening socket. + match io::Error::last_os_error().raw_os_error() { + Some(libc::ENOPROTOOPT) | Some(libc::EOPNOTSUPP) => None, + _ => Some(false), + } +} + +#[cfg(not(any(target_os = "linux", target_os = "android")))] +fn socket_type(fd: RawFd) -> Option { + let mut val: libc::c_int = 0; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: `val`/`len` are live locals of the sizes the option expects. + let rc = unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_TYPE, + &mut val as *mut libc::c_int as *mut c_void, + &mut len, + ) + }; + (rc == 0).then_some(val) +} + +#[cfg(not(any(target_os = "linux", target_os = "android")))] +fn socket_is_connected(fd: RawFd) -> bool { + let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() }; + let mut len = size_of::() as libc::socklen_t; + // SAFETY: `storage` is a live, correctly-sized `sockaddr_storage` and + // `len` describes it accurately. + let rc = unsafe { + libc::getpeername( + fd, + &mut storage as *mut libc::sockaddr_storage as *mut libc::sockaddr, + &mut len, + ) + }; + rc == 0 +} + +/// Prepare a control-socket endpoint: close-on-exec, SIGPIPE suppression +/// where it is a socket option, and a bounded write timeout. +/// +/// The write timeout matters as much as the read timeouts the protocol code +/// already arms. Without `SO_SNDTIMEO`, a peer that stops reading (stuck in a +/// consumer hook, or `SIGSTOP`ped) lets our socket buffer fill and parks the +/// writer in an uninterruptible `write(2)` forever — the one place in the +/// handoff where neither the liveness clock nor the overall deadline is +/// running, because both are enforced on the read side. +pub fn configure_control_stream(stream: &UnixStream, write_timeout: Duration) -> io::Result<()> { + set_cloexec(stream.as_raw_fd())?; + set_nosigpipe(stream.as_raw_fd()); + // A closed peer makes `setsockopt` return EINVAL on macOS/BSD (see + // `supervisor::arm_recv_timeout` for the full story). A write to that + // socket cannot block, so the missing timeout is harmless. + match stream.set_write_timeout(Some(write_timeout)) { + Ok(()) => Ok(()), + Err(e) if e.raw_os_error() == Some(libc::EINVAL) => Ok(()), + Err(e) => Err(e), + } +} + +/// Ask the kernel to report `EPIPE` instead of raising `SIGPIPE` on this +/// socket. Only meaningful on platforms with the option; Linux uses the +/// per-call `MSG_NOSIGNAL` flag in [`send_all`] instead. Best-effort: a +/// failure here leaves the process's SIGPIPE disposition in charge. +#[allow(unused_variables)] +fn set_nosigpipe(fd: RawFd) { + #[cfg(any( + target_os = "macos", + target_os = "ios", + target_os = "freebsd", + target_os = "netbsd", + target_os = "openbsd", + target_os = "dragonfly" + ))] + { + let val: libc::c_int = 1; + // SAFETY: `val` is a live `c_int` and the length matches it. + unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_NOSIGPIPE, + &val as *const libc::c_int as *const c_void, + size_of::() as libc::socklen_t, + ); + } + } +} + +/// Write `buf` to `stream` in full, never raising `SIGPIPE`. +/// +/// Rust's runtime ignores `SIGPIPE` by default, but that is a property of the +/// *binary*, not of this library: a consumer that restores the default +/// disposition (common in CLI-shaped daemons that want to die politely when +/// piped into `head`) would otherwise be killed outright when a peer dies +/// mid-handoff. Suppressing the signal here makes the failure an ordinary +/// `EPIPE` that the protocol code already knows how to handle. +pub(crate) fn send_all(stream: &UnixStream, buf: &[u8]) -> io::Result<()> { + #[cfg(any(target_os = "linux", target_os = "android"))] + const FLAGS: libc::c_int = libc::MSG_NOSIGNAL; + #[cfg(not(any(target_os = "linux", target_os = "android")))] + const FLAGS: libc::c_int = 0; + + let mut sent = 0usize; + while sent < buf.len() { + // SAFETY: `buf[sent..]` is a live slice; `send` writes at most + // `len` bytes from it and never retains the pointer. + let n = unsafe { + libc::send( + stream.as_raw_fd(), + buf[sent..].as_ptr() as *const c_void, + buf.len() - sent, + FLAGS, + ) + }; + if n < 0 { + let err = io::Error::last_os_error(); + if err.kind() == io::ErrorKind::Interrupted { + continue; + } + return Err(err); + } + if n == 0 { + return Err(io::Error::from(io::ErrorKind::WriteZero)); + } + sent += n as usize; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_over_long_socket_path() { + let long = PathBuf::from("/tmp").join("x".repeat(max_socket_path_len())); + match validate_socket_path(&long) { + Err(Error::SocketPathTooLong { len, max, .. }) => { + assert!(len > max); + assert_eq!(max, max_socket_path_len()); + } + other => panic!("expected SocketPathTooLong, got {other:?}"), + } + } + + #[test] + fn bind_produces_owner_only_socket() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("ctl.sock"); + let listener = bind_socket(&path).unwrap(); + let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777; + assert_eq!( + mode, SOCKET_MODE, + "socket must not be group/world reachable" + ); + // No staging files left behind. + let leftovers: Vec<_> = std::fs::read_dir(dir.path()) + .unwrap() + .filter_map(|e| e.ok()) + .filter(|e| e.file_name().to_string_lossy().ends_with(".tmp")) + .collect(); + assert!(leftovers.is_empty(), "staging file leaked: {leftovers:?}"); + drop(listener); + } + + #[test] + fn rebind_replaces_path_without_a_missing_window() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("ctl.sock"); + let first = bind_socket(&path).unwrap(); + let first_ino = inode_of(&path); + // The second bind takes over the name atomically; the path exists + // continuously and now refers to the new socket. + let second = bind_socket(&path).unwrap(); + assert!(path.exists()); + assert_ne!(first_ino, inode_of(&path)); + // The displaced listener is still open (its binding is just + // unreachable by name), which is what lets the prior incumbent keep + // serving already-accepted sessions. + assert!(fd_is_listening(first.as_raw_fd())); + drop(second); + } + + #[test] + fn cloexec_and_blocking_normalization_are_idempotent() { + let dir = tempfile::tempdir().unwrap(); + let listener = bind_socket(&dir.path().join("s.sock")).unwrap(); + let fd = listener.as_raw_fd(); + listener.set_nonblocking(true).unwrap(); + for _ in 0..2 { + clear_nonblocking(fd).unwrap(); + set_cloexec(fd).unwrap(); + } + // SAFETY: probing flags on a live fd we own. + let fl = unsafe { libc::fcntl(fd, libc::F_GETFL) }; + let fd_flags = unsafe { libc::fcntl(fd, libc::F_GETFD) }; + assert_eq!(fl & libc::O_NONBLOCK, 0); + assert_ne!(fd_flags & libc::FD_CLOEXEC, 0); + } + + #[test] + fn socket_introspection_distinguishes_kinds() { + let dir = tempfile::tempdir().unwrap(); + let listener = bind_socket(&dir.path().join("s.sock")).unwrap(); + assert!(fd_is_socket(listener.as_raw_fd())); + assert!(fd_is_listening(listener.as_raw_fd())); + assert_eq!( + socket_family(listener.as_raw_fd()), + Some(libc::AF_UNIX as libc::sa_family_t) + ); + + let file = std::fs::File::create(dir.path().join("plain")).unwrap(); + assert!(!fd_is_socket(file.as_raw_fd())); + assert!(!fd_is_listening(file.as_raw_fd())); + + let conn = UnixStream::connect(dir.path().join("s.sock")).unwrap(); + assert!(fd_is_socket(conn.as_raw_fd())); + assert!(!fd_is_listening(conn.as_raw_fd())); + } + + #[test] + fn peer_uid_matches_our_own_for_a_local_connection() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("s.sock"); + let listener = bind_socket(&path).unwrap(); + let _client = UnixStream::connect(&path).unwrap(); + let (server, _) = listener.accept().unwrap(); + // SAFETY: `geteuid` cannot fail. + let euid = unsafe { libc::geteuid() }; + assert_eq!(peer_uid(&server).unwrap(), euid); + assert!(peer_uid_is_allowed(euid, &[])); + assert!(!peer_uid_is_allowed(euid.wrapping_add(1), &[])); + assert!(peer_uid_is_allowed(euid.wrapping_add(1), &[euid + 1])); + } + + fn inode_of(path: &Path) -> u64 { + use std::os::unix::fs::MetadataExt; + std::fs::metadata(path).unwrap().ino() + } +} diff --git a/crates/handoff/src/supervisor.rs b/crates/handoff/src/supervisor.rs index 4cb3f4e..e563e6f 100644 --- a/crates/handoff/src/supervisor.rs +++ b/crates/handoff/src/supervisor.rs @@ -25,11 +25,12 @@ use crate::crash::points; use crate::crash_here; use crate::error::{Error, Result}; use crate::fd::pass_listener_fds_on_spawn; -use crate::frame::{read_message, write_message}; +use crate::frame::{FrameAccumulator, read_message, write_frame}; use crate::metrics::events; use crate::protocol::{ HandoffId, Message, PROTO_MAX, PROTO_MIN, ProtoVersion, Side, negotiate_version, short_name, }; +use crate::sock; use crate::state::{Phase, StateJournal}; use crate::util::now_unix_ms; @@ -194,7 +195,11 @@ impl Drop for ChildGuard { } impl Supervisor { + /// Fails with [`Error::SocketPathTooLong`] if `socket_path` cannot fit in + /// `sockaddr_un` — better here, at construction, than as an opaque + /// `EINVAL` from `connect(2)` in the middle of a handoff. pub fn new(socket_path: &Path) -> Result { + sock::validate_socket_path(socket_path)?; Ok(Self { socket_path: socket_path.to_path_buf(), listener_fds: Vec::new(), @@ -242,8 +247,20 @@ impl Supervisor { // 1. Connect to O. let mut o_stream = UnixStream::connect(&self.socket_path)?; - let chosen_o = - self.exchange_hello_as_supervisor(&mut o_stream, handoff_id, Side::Incumbent, None)?; + sock::configure_control_stream(&o_stream, sock::CONTROL_WRITE_TIMEOUT)?; + // One accumulator per stream, held for the whole handoff: it carries + // partial frames across the per-phase receive timeouts that are armed + // and disarmed below, so a reply split by a timeout boundary resumes + // instead of desynchronizing the stream. + let mut o_acc = FrameAccumulator::new(); + let mut n_acc = FrameAccumulator::new(); + let chosen_o = self.exchange_hello_as_supervisor( + &mut o_stream, + &mut o_acc, + handoff_id, + Side::Incumbent, + None, + )?; crash_here!(points::S_AFTER_O_HELLO); // 2. Create a socketpair for N's control channel. @@ -260,10 +277,12 @@ impl Supervisor { crash_here!(points::S_AFTER_SPAWN_SUCCESSOR); let mut n_stream = s_end; + sock::configure_control_stream(&n_stream, sock::CONTROL_WRITE_TIMEOUT)?; // 4. Hello/HelloAck with N. Verify the child's announced PID matches // the one we spawned. let chosen_n = self.exchange_hello_as_supervisor( &mut n_stream, + &mut n_acc, handoff_id, Side::Successor, Some(successor_pid), @@ -293,8 +312,8 @@ impl Supervisor { deadline_ms, "prepare handoff" ); - write_message( - &mut o_stream, + write_frame( + &o_stream, chosen_o, &Message::PrepareHandoff { handoff_id, @@ -316,6 +335,7 @@ impl Supervisor { // reply that's already on the wire. let drained_msg = read_until( &mut o_stream, + &mut o_acc, spec.drain_grace + WIRE_SLACK, "Drained", |m| matches!(m, Message::Drained { .. }), @@ -341,11 +361,7 @@ impl Supervisor { // 6. SealRequest → SealComplete (or SealFailed). let seal_at = Instant::now(); tracing::info!(target: events::SEAL, %handoff_id, "seal request"); - write_message( - &mut o_stream, - chosen_o, - &Message::SealRequest { handoff_id }, - )?; + write_frame(&o_stream, chosen_o, &Message::SealRequest { handoff_id })?; crash_here!(points::S_AFTER_SEAL_REQUEST_SENT); // Seal-wait loop. Same two-tier timeout as `read_until`: per-recv // capped at LIVENESS_TIMEOUT (heartbeats reset it), wall-clock @@ -372,28 +388,33 @@ impl Supervisor { let remaining = seal_read_deadline - now; let recv_timeout = LIVENESS_TIMEOUT.min(remaining).max(MIN_READ_TIMEOUT); arm_recv_timeout(&o_stream, recv_timeout)?; - match read_message(&mut o_stream) { - Ok((_, Message::SealProgress { .. })) => continue, - Ok((_, Message::Heartbeat { .. })) => continue, - Ok((_, Message::SealComplete { handoff_id: id, .. })) if id == handoff_id => { + match o_acc.poll_read(&mut o_stream) { + Ok(Some((_, Message::SealProgress { .. }))) => continue, + Ok(Some((_, Message::Heartbeat { .. }))) => continue, + Ok(Some((_, Message::SealComplete { handoff_id: id, .. }))) if id == handoff_id => { break Ok(()); } - Ok(( + Ok(Some(( _, Message::SealFailed { handoff_id: id, error, .. }, - )) if id == handoff_id => break Err(error), - Ok((_, other)) => { + ))) if id == handoff_id => break Err(error), + Ok(Some((_, other))) => { let _ = o_stream.set_read_timeout(None); return Err(Error::UnexpectedMessage(short_name(&other))); } - Err(Error::Io(e)) if is_timeout(&e) => { - // No frame for `recv_timeout` — peer has gone silent - // for longer than LIVENESS_TIMEOUT (or we're at the - // overall wall-clock cap; the next loop iteration + // Timed out mid-frame: O is demonstrably alive (it is + // writing), so this is not the peer-dead condition. Keep + // reading — the wall-clock check at the top of the loop still + // bounds the phase. + Ok(None) if o_acc.has_partial() => continue, + Ok(None) => { + // No bytes at all for `recv_timeout` — peer has gone + // silent for longer than LIVENESS_TIMEOUT (or we're at + // the overall wall-clock cap; the next loop iteration // detects that explicitly). Treat as peer-dead and // abort. let _ = o_stream.set_read_timeout(None); @@ -453,28 +474,27 @@ impl Supervisor { // correct response in both cases (N is unreachable; O has sealed // and must be told to keep serving). let begin_at = Instant::now(); - let ready_result = - match write_message(&mut n_stream, chosen_n, &Message::Begin { handoff_id }) { - Ok(()) => { - crash_here!(points::S_AFTER_BEGIN_SENT); - self.journal_set( - handoff_id, - Phase::AwaitingReady, - successor_pid, - started_unix_ms, - )?; - // N has no internal deadline for `announce_and_bind`, - // so `Ready` can be in flight at the moment - // `total_deadline_at` elapses — give the read - // `WIRE_SLACK` past that cap for the same reason the - // drain and seal reads do. - let ready_timeout = remaining_until(total_deadline_at) + WIRE_SLACK; - read_until(&mut n_stream, ready_timeout, "Ready", |m| { - matches!(m, Message::Ready { .. }) - }) - } - Err(e) => Err(e), - }; + let ready_result = match write_frame(&n_stream, chosen_n, &Message::Begin { handoff_id }) { + Ok(()) => { + crash_here!(points::S_AFTER_BEGIN_SENT); + self.journal_set( + handoff_id, + Phase::AwaitingReady, + successor_pid, + started_unix_ms, + )?; + // N has no internal deadline for `announce_and_bind`, + // so `Ready` can be in flight at the moment + // `total_deadline_at` elapses — give the read + // `WIRE_SLACK` past that cap for the same reason the + // drain and seal reads do. + let ready_timeout = remaining_until(total_deadline_at) + WIRE_SLACK; + read_until(&mut n_stream, &mut n_acc, ready_timeout, "Ready", |m| { + matches!(m, Message::Ready { .. }) + }) + } + Err(e) => Err(e), + }; // The `Ready` match arm pulls the handoff_id apart so a mismatched id // produces a precise error rather than the misleading @@ -523,9 +543,7 @@ impl Supervisor { total_seconds = started_instant.elapsed().as_secs_f64(), "commit" ); - if let Err(e) = - write_message(&mut o_stream, chosen_o, &Message::Commit { handoff_id }) - { + if let Err(e) = write_frame(&o_stream, chosen_o, &Message::Commit { handoff_id }) { tracing::warn!( %handoff_id, error = %e, "failed to send Commit to incumbent; O may have crashed — \ @@ -564,8 +582,8 @@ impl Supervisor { // Abort N, resume O. send_best_effort_abort(&mut n_stream, chosen_n, handoff_id, reason.clone()); child_guard.kill_and_reap(); - write_message( - &mut o_stream, + write_frame( + &o_stream, chosen_o, &Message::ResumeAfterAbort { handoff_id }, )?; @@ -645,6 +663,7 @@ impl Supervisor { fn exchange_hello_as_supervisor( &self, stream: &mut UnixStream, + acc: &mut FrameAccumulator, handoff_id: HandoffId, expected_role: Side, expected_pid: Option, @@ -653,11 +672,11 @@ impl Supervisor { // hangs before writing can't block `perform_handoff` indefinitely. // Cleared after the read regardless of outcome. arm_recv_timeout(stream, HELLO_READ_TIMEOUT)?; - let read_result = read_message(stream); + let read_result = acc.poll_read(stream); let _ = stream.set_read_timeout(None); let (_v, peer_hello) = match read_result { - Ok(x) => x, - Err(Error::Io(e)) if is_timeout(&e) => return Err(Error::Timeout("peer Hello")), + Ok(Some(x)) => x, + Ok(None) => return Err(Error::Timeout("peer Hello")), Err(e) => return Err(e), }; let (their_role, their_pid, their_min, their_max) = match peer_hello { @@ -685,7 +704,7 @@ impl Supervisor { }); } let chosen = negotiate_version(PROTO_MIN, PROTO_MAX, their_min, their_max)?; - write_message( + write_frame( stream, chosen, &Message::HelloAck { @@ -764,7 +783,7 @@ fn send_best_effort_abort( reason: String, ) { let reason_for_log = reason.clone(); - if let Err(e) = write_message(stream, version, &Message::Abort { handoff_id, reason }) { + if let Err(e) = write_frame(stream, version, &Message::Abort { handoff_id, reason }) { tracing::warn!( %handoff_id, reason = %reason_for_log, @@ -837,6 +856,7 @@ fn make_socketpair() -> Result<(UnixStream, UnixStream)> { /// step expired (e.g. `Timeout("Drained")` vs `Timeout("Ready")`). fn read_until( stream: &mut UnixStream, + acc: &mut FrameAccumulator, timeout: Duration, awaiting: &'static str, pred: F, @@ -858,18 +878,22 @@ where // the next iteration. let recv_timeout = LIVENESS_TIMEOUT.min(remaining).max(MIN_READ_TIMEOUT); arm_recv_timeout(stream, recv_timeout)?; - match read_message(stream) { - Ok((_, Message::Heartbeat { .. })) => continue, - Ok((_, Message::SealProgress { .. })) => continue, - Ok((_, msg)) if pred(&msg) => { + match acc.poll_read(stream) { + Ok(Some((_, Message::Heartbeat { .. }))) => continue, + Ok(Some((_, Message::SealProgress { .. }))) => continue, + Ok(Some((_, msg))) if pred(&msg) => { let _ = stream.set_read_timeout(None); return Ok(msg); } - Ok((_, other)) => { + Ok(Some((_, other))) => { let _ = stream.set_read_timeout(None); return Err(Error::UnexpectedMessage(short_name(&other))); } - Err(Error::Io(e)) if is_timeout(&e) => { + // Mid-frame at the timeout: the peer is writing, so the liveness + // clock has no business firing. The wall-clock deadline above + // still bounds the wait. + Ok(None) if acc.has_partial() => continue, + Ok(None) => { let _ = stream.set_read_timeout(None); return Err(Error::Timeout(awaiting)); } @@ -935,7 +959,3 @@ fn remaining_until(deadline: Instant) -> Duration { .unwrap_or(MIN_READ_TIMEOUT) .max(MIN_READ_TIMEOUT) } - -fn is_timeout(e: &std::io::Error) -> bool { - matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) -}