fix: fail the launch when the egress socket cannot be bound

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
 

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::<libc::sockaddr_un>() - std::mem::offset_of!(libc::sockaddr_un, sun_path);
+
 struct Inner {
     memory: Arc<Memory>,
     destinations: HashMap<String, Destination>,
@@ -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<tokio::net::UnixListener, Error> {
+        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<L>(&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();

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<String>,
+    /// Serving the egress socket, for every runner but the unconfined one.
     #[cfg(unix)]
-    serve_unix: tokio::task::JoinHandle<Result<(), dm_egress::Error>>,
+    serve_unix: Option<tokio::task::JoinHandle<Result<(), dm_egress::Error>>>,
     /// 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<status::Writer>,
@@ -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;