diff --git a/Cargo.lock b/Cargo.lock index 4861e27..131a645 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -898,6 +898,7 @@ dependencies = [ "futures-util", "hyper", "hyper-util", + "libc", "opentelemetry", "opentelemetry_sdk", "reqwest 0.13.1", diff --git a/crates/egress/Cargo.toml b/crates/egress/Cargo.toml index 5e5ecec..2f448bc 100644 --- a/crates/egress/Cargo.toml +++ b/crates/egress/Cargo.toml @@ -25,6 +25,9 @@ thiserror.workspace = true time = { version = "0.3.45", features = ["formatting", "parsing"] } tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-util", "sync"] } +[target.'cfg(unix)'.dependencies] +libc = "0.2.189" + [lints] workspace = true diff --git a/crates/egress/src/lib.rs b/crates/egress/src/lib.rs index f12ca30..9694ac1 100644 --- a/crates/egress/src/lib.rs +++ b/crates/egress/src/lib.rs @@ -194,11 +194,31 @@ pub enum Error { /// A direct call the upstream did not answer with success. #[error("{0}")] Upstream(String), + /// A unix socket path longer than the system takes. + #[error( + "egress socket {} is {len} bytes long, and a unix socket path takes at most {max} bytes here, {} with the closing NUL", + .path.display(), + .max + 1 + )] + SocketPath { + /// The path. + path: std::path::PathBuf, + /// Its length in bytes. + len: usize, + /// The longest path the system takes, in bytes, without the closing NUL. + max: usize, + }, /// A tool server named for a tool the grants do not host. #[error("{0} is not a hosted tool in grants.toml; a tool server is reached only for [taps.{0}] hosted = \"...\"")] NotHosted(String), } +/// `sizeof(sockaddr_un.sun_path)`: the most a unix socket path takes, its closing NUL +/// included. The field is the struct's last, so its size is what follows its offset. +#[cfg(unix)] +pub const SUN_PATH: usize = + std::mem::size_of::() - std::mem::offset_of!(libc::sockaddr_un, sun_path); + struct Inner { memory: Arc, destinations: HashMap, @@ -560,18 +580,57 @@ impl Egress { .with_state(self.clone()) } - /// Serve on a unix socket until the task is cancelled. Removes a stale socket first. - /// Unix only; Windows runs the harness unconfined over TCP. + /// Bind the unix socket the egress serves on. A path longer than the system takes is + /// refused with its length. A socket already at the path is removed when no listener + /// holds it, and refused when one does; anything else there is refused and left in + /// place. Needs a Tokio runtime. Unix only; Windows runs the harness unconfined over + /// TCP. + /// + /// # Errors + /// + /// When the path is too long, something other than a stale socket is there, or the + /// socket cannot be bound. + #[cfg(unix)] + pub fn bind_unix(path: &FsPath) -> Result { + let len = path.as_os_str().len(); + if len >= SUN_PATH { + return Err(Error::SocketPath { + path: path.to_path_buf(), + len, + max: SUN_PATH - 1, + }); + } + let named = |e: std::io::Error| { + Error::Io(std::io::Error::new( + e.kind(), + format!("bind {}: {e}", path.display()), + )) + }; + remove_stale_socket(path).map_err(named)?; + tokio::net::UnixListener::bind(path).map_err(named) + } + + /// Serve on a unix socket until the task is cancelled, as [`Egress::bind_unix`] binds + /// it. /// /// # Errors /// /// When the socket cannot be bound or serving fails. #[cfg(unix)] pub async fn serve_unix(&self, path: &FsPath) -> Result<(), Error> { - if path.exists() { - std::fs::remove_file(path)?; - } - let listener = tokio::net::UnixListener::bind(path)?; + self.serve(Self::bind_unix(path)?).await + } + + /// Serve on a listener already bound, unix or TCP, until the task is cancelled. + /// + /// # Errors + /// + /// When serving fails. + pub async fn serve(&self, listener: L) -> Result<(), Error> + where + L: axum::serve::Listener, + L::Addr: std::fmt::Debug, + { axum::serve(listener, self.router()).await?; Ok(()) } @@ -582,9 +641,35 @@ impl Egress { /// /// When the address cannot be bound or serving fails. pub async fn serve_tcp(&self, addr: std::net::SocketAddr) -> Result<(), Error> { - let listener = tokio::net::TcpListener::bind(addr).await?; - axum::serve(listener, self.router()).await?; - Ok(()) + self.serve(tokio::net::TcpListener::bind(addr).await?).await + } +} + +/// Remove a socket no listener holds. Nothing there is fine; a socket a listener answers +/// on, or anything that is not a socket, is an error and stays. +#[cfg(unix)] +fn remove_stale_socket(path: &FsPath) -> std::io::Result<()> { + use std::io::ErrorKind; + use std::os::unix::fs::FileTypeExt; + let meta = match std::fs::symlink_metadata(path) { + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(()), + other => other?, + }; + if !meta.file_type().is_socket() { + return Err(std::io::Error::new( + ErrorKind::AlreadyExists, + "something that is not a socket is there; it is left in place", + )); + } + match std::os::unix::net::UnixStream::connect(path) { + Ok(_) => Err(std::io::Error::new( + ErrorKind::AddrInUse, + "the socket is in use by another launch", + )), + Err(e) if matches!(e.kind(), ErrorKind::ConnectionRefused | ErrorKind::NotFound) => { + std::fs::remove_file(path) + } + Err(e) => Err(e), } } @@ -1541,6 +1626,94 @@ mod tests { assert!(text.contains("\"serverInfo\""), "{text}"); } + /// A path under `dir` of exactly `len` bytes, its directories made. + #[cfg(unix)] + fn socket_path_of(dir: &FsPath, len: usize) -> std::path::PathBuf { + let base = dir.join("egress.sock").as_os_str().len(); + let pad = dir.join("p".repeat(len - base - 1)); + std::fs::create_dir_all(&pad).unwrap(); + let path = pad.join("egress.sock"); + assert_eq!(path.as_os_str().len(), len); + path + } + + /// `sizeof(sockaddr_un.sun_path)` as the system headers give it, not as the egress + /// does. + #[cfg(unix)] + const SUN_PATH: usize = if cfg!(any(target_os = "linux", target_os = "android")) { + 108 + } else { + 104 + }; + + #[cfg(unix)] + #[tokio::test] + async fn a_socket_path_too_long_for_the_system_is_refused_with_its_length() { + let (d, e) = fixture(None).await; + let path = socket_path_of(d.path(), SUN_PATH); + let err = e.serve_unix(&path).await.unwrap_err().to_string(); + assert!(err.contains(&path.display().to_string()), "{err}"); + assert!(err.contains(&format!("{SUN_PATH} bytes")), "{err}"); + assert!(err.contains(&format!("at most {}", SUN_PATH - 1)), "{err}"); + assert!(!path.exists()); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_socket_path_one_byte_under_the_limit_is_served() { + let (d, e) = fixture(None).await; + let path = socket_path_of(d.path(), SUN_PATH - 1); + let serving = + tokio::time::timeout(std::time::Duration::from_millis(200), e.serve_unix(&path)).await; + assert!(serving.is_err(), "still serving: {serving:?}"); + assert!(path.exists()); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_socket_no_listener_holds_is_replaced() { + let d = tempfile::tempdir().unwrap(); + let path = d.path().join("egress.sock"); + drop(Egress::bind_unix(&path).unwrap()); + assert!(path.exists(), "a dropped listener leaves its file behind"); + let again = Egress::bind_unix(&path).unwrap(); + let _client = std::os::unix::net::UnixStream::connect(&path).unwrap(); + again.accept().await.unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_socket_another_listener_holds_is_left_to_it() { + let d = tempfile::tempdir().unwrap(); + let path = d.path().join("egress.sock"); + let held = Egress::bind_unix(&path).unwrap(); + let err = Egress::bind_unix(&path).unwrap_err().to_string(); + assert!(err.contains(&path.display().to_string()), "{err}"); + assert!(err.contains("in use by another launch"), "{err}"); + let _client = std::os::unix::net::UnixStream::connect(&path).unwrap(); + held.accept().await.unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn a_path_that_is_not_a_socket_is_refused_and_left_in_place() { + let d = tempfile::tempdir().unwrap(); + let file = d.path().join("egress.sock"); + std::fs::write(&file, "kept").unwrap(); + let err = Egress::bind_unix(&file).unwrap_err().to_string(); + assert!(err.contains(&file.display().to_string()), "{err}"); + assert!(err.contains("not a socket"), "{err}"); + assert_eq!(std::fs::read_to_string(&file).unwrap(), "kept"); + + let link = d.path().join("link.sock"); + std::os::unix::fs::symlink(&file, &link).unwrap(); + let err = Egress::bind_unix(&link).unwrap_err().to_string(); + assert!(err.contains(&link.display().to_string()), "{err}"); + assert!(err.contains("not a socket"), "{err}"); + assert!(link.symlink_metadata().unwrap().file_type().is_symlink()); + assert_eq!(std::fs::read_to_string(&file).unwrap(), "kept"); + } + #[tokio::test] async fn a_destination_without_a_ceiling_is_refused_at_build_time() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/memory/src/main.rs b/crates/memory/src/main.rs index a94484c..9e96d2f 100644 --- a/crates/memory/src/main.rs +++ b/crates/memory/src/main.rs @@ -3395,8 +3395,9 @@ pub struct Session { /// The model, in the harness's own flag, after every run's arguments: a flag after /// the prompt is where every harness, subcommand or not, accepts one. model: Option, + /// Serving the egress socket, for every runner but the unconfined one. #[cfg(unix)] - serve_unix: tokio::task::JoinHandle>, + serve_unix: Option>>, /// What the session says about itself to the web launcher that started it. #[allow(dead_code)] // kept for its drop, which writes the last word status: Option, @@ -3482,19 +3483,16 @@ impl Session { p.home.clone_from(h); } + // bound here, so a socket that cannot be bound fails the launch with its own error; + // the unconfined harness reaches the egress over TCP and needs no socket #[cfg(unix)] - let serve_unix = { + let serve_unix = if matches!(runner, Runner::Unconfined(_)) { + None + } else { + let listener = Egress::bind_unix(&p.socket)?; let egress = p.egress.clone(); - let socket = p.socket.clone(); - tokio::spawn(async move { egress.serve_unix(&socket).await }) + Some(tokio::spawn(async move { egress.serve(listener).await })) }; - #[cfg(unix)] - for _ in 0..100 { - if p.socket.exists() { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } if let Runner::Requests(r) = &runner { // the harness container runs as uid 1000 and must reach the socket and leave // its exit code beside it; this process is root in the egress container @@ -3514,11 +3512,10 @@ impl Session { if let Runner::Unconfined(slot) = &mut runner { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?; - drop(listener); let egress = p.egress.clone(); *slot = Some(( addr, - tokio::spawn(async move { egress.serve_tcp(addr).await }), + tokio::spawn(async move { egress.serve(listener).await }), )); } announce(&runner, &p, harness, &workdir); @@ -3818,8 +3815,8 @@ impl Session { serve.abort(); } #[cfg(unix)] - { - self.serve_unix.abort(); + if let Some(serve) = &self.serve_unix { + serve.abort(); std::fs::remove_file(&self.p.socket).ok(); } if !self.p.metered { @@ -3945,12 +3942,11 @@ fn relay( .enable_all() .build()?; let socket = socket.to_path_buf(); - let listen = listen.to_string(); + // bound before the harness starts, which without it has no way out + let listener = rt + .block_on(tokio::net::TcpListener::bind(listen)) + .map_err(|e| format!("cannot listen on {listen}: {e}"))?; rt.spawn(async move { - let Ok(listener) = tokio::net::TcpListener::bind(&listen).await else { - eprintln!("relay: cannot listen on {listen}"); - return; - }; loop { let Ok((mut inbound, _)) = listener.accept().await else { continue; diff --git a/crates/memory/tests/launch.rs b/crates/memory/tests/launch.rs index a3d50c7..722d87d 100644 --- a/crates/memory/tests/launch.rs +++ b/crates/memory/tests/launch.rs @@ -253,6 +253,16 @@ fn confined_launch_has_no_network_and_no_host_home() { assert!(text.contains("HOME=/home/user"), "{text}"); assert!(text.contains(&format!("cwd={}", work.display())), "{text}"); assert!(stderr.contains("confined to"), "{stderr}"); + let left: Vec<_> = std::fs::read_dir(root.join("state/run")) + .unwrap() + .flatten() + .map(|e| e.file_name().to_string_lossy().into_owned()) + .filter(|n| n.starts_with("egress-")) + .collect(); + assert!( + left.is_empty(), + "the session takes its socket away: {left:?}" + ); } #[cfg(target_os = "linux")] @@ -335,6 +345,156 @@ fn a_launch_that_fails_after_rendering_leaves_no_render_behind() { assert!(left.is_empty(), "a render outlived its session: {left:?}"); } +/// A root whose egress socket, `state/run/egress-.sock`, is longer than any system +/// lets a unix socket path be. +fn root_too_long_for_its_socket(dir: &Path) -> PathBuf { + let deep = dir.join("d".repeat(120)); + std::fs::create_dir_all(&deep).unwrap(); + init_root(&deep) +} + +#[cfg(target_os = "linux")] +#[test] +fn a_confined_launch_whose_socket_cannot_be_bound_fails_and_says_why() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = root_too_long_for_its_socket(dir.path()); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + let bins = fake_claude(&work); + + let out = launch(&root, &bins, &work, &[]); + let stderr = String::from_utf8_lossy(&out.stderr); + assert_eq!(out.status.code(), Some(1), "{stderr}"); + assert!( + !work.join("fake-claude.out").exists(), + "the harness did not start: {stderr}" + ); + let prefix = root.join("state/run/egress-").display().to_string(); + let at = stderr.find(&prefix).expect(&stderr); + let end = stderr[at..].find(".sock").expect(&stderr) + ".sock".len(); + let socket = &stderr[at..at + end]; + assert!( + stderr.contains(&format!("{} bytes", socket.len())), + "the error names the socket's length: {stderr}" + ); +} + +#[test] +fn an_unconfined_launch_needs_no_socket_and_runs_from_a_long_root() { + let dir = tempfile::tempdir().unwrap(); + let root = root_too_long_for_its_socket(dir.path()); + let bins = fake_claude(dir.path()); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + + let out = launch(&root, &bins, &work, &["--unconfined"]); + assert_eq!( + out.status.code(), + Some(42), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + assert!(read_out(&work).contains("mcp=200"), "{}", read_out(&work)); +} + +#[test] +fn an_unconfined_launch_binds_no_socket() { + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let run = root.join("state/run"); + let bins = fake_harness( + dir.path(), + &format!("ls -a '{}' > ./run.ls; exit 0", run.display()), + ); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + + let out = launch(&root, &bins, &work, &["--unconfined"]); + let stderr = String::from_utf8_lossy(&out.stderr); + assert_eq!(out.status.code(), Some(0), "{stderr}"); + let seen = std::fs::read_to_string(work.join("run.ls")).unwrap(); + assert!( + !seen.contains("egress-"), + "state/run while the harness ran: {seen}" + ); +} + +/// `amem inside` with this requests directory, a subscription login in a host home, and +/// a prompt; the caller answers the requests. +fn inside(root: &Path, requests: &Path, host_home: &Path) -> std::process::Child { + std::fs::create_dir_all(host_home.join(".claude")).unwrap(); + std::fs::write(host_home.join(".claude/.credentials.json"), "{}").unwrap(); + memory() + .args([ + "--root", + root.to_str().unwrap(), + "inside", + "--harness", + "claude", + "--auth", + "subscription", + ]) + .args([ + "--requests", + requests.to_str().unwrap(), + "--host-home", + host_home.to_str().unwrap(), + ]) + .args(["--prompt", "say the answer", "--no-handoff"]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap() +} + +#[test] +fn inside_with_a_requests_directory_too_long_for_its_socket_fails_and_says_why() { + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let requests = dir.path().join("v".repeat(120)); + std::fs::create_dir_all(&requests).unwrap(); + let socket = requests.join("egress.sock"); + + let out = inside(&root, &requests, &dir.path().join("host-home")) + .wait_with_output() + .unwrap(); + let stderr = String::from_utf8_lossy(&out.stderr); + assert_eq!(out.status.code(), Some(1), "{stderr}"); + assert!( + stderr.contains(&format!("{} bytes", socket.as_os_str().len())), + "the error names the socket's length: {stderr}" + ); + assert!( + !requests.join("req-0.json").exists(), + "no run was asked for" + ); +} + +#[test] +fn a_relay_that_cannot_listen_does_not_run_the_harness() { + let dir = tempfile::tempdir().unwrap(); + let taken = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listen = taken.local_addr().unwrap().to_string(); + let marker = dir.path().join("ran"); + let out = memory() + .args(["relay", "--socket"]) + .arg(dir.path().join("egress.sock")) + .args(["--listen", &listen, "--", "touch"]) + .arg(&marker) + .stdin(Stdio::null()) + .output() + .unwrap(); + let stderr = String::from_utf8_lossy(&out.stderr); + assert_eq!(out.status.code(), Some(1), "{stderr}"); + assert!(!marker.exists(), "the harness ran without a route out"); + assert!(stderr.contains(&listen), "{stderr}"); +} + // ---- non-interactive: a prompt, and a task from the ledger ---- /// A fake harness from a shell body. It receives the launcher's arguments; `$*` holds @@ -743,32 +903,8 @@ fn inside_writes_image_requests_and_reports_done() { let root = init_root(dir.path()); let requests = dir.path().join("volume"); std::fs::create_dir_all(&requests).unwrap(); - let host_home = dir.path().join("host-home"); - std::fs::create_dir_all(host_home.join(".claude")).unwrap(); - std::fs::write(host_home.join(".claude/.credentials.json"), "{}").unwrap(); - let child = memory() - .args([ - "--root", - root.to_str().unwrap(), - "inside", - "--harness", - "claude", - "--auth", - "subscription", - ]) - .args([ - "--requests", - requests.to_str().unwrap(), - "--host-home", - host_home.to_str().unwrap(), - ]) - .args(["--prompt", "say the answer", "--no-handoff"]) - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .unwrap(); + let child = inside(&root, &requests, &dir.path().join("host-home")); let req = await_request(&requests.join("req-0.json")); check_image_request(&req); @@ -793,6 +929,10 @@ fn inside_writes_image_requests_and_reports_done() { .trim(), "0" ); + assert!( + !requests.join("egress.sock").exists(), + "the session takes its socket away with it" + ); } // ---- hooks: a handoff the harness did not write ----