windows: Add support for SSH (#29145)

Closes #19892

This PR builds on top of #20587 and improves upon it.

Release Notes:

- N/A

---------

Co-authored-by: Kirill Bulatov <kirill@zed.dev>

crates/net/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "net"
+version = "0.1.0"
+edition.workspace = true
+publish.workspace = true
+license = "GPL-3.0-or-later"
+
+[lints]
+workspace = true
+
+[lib]
+path = "src/net.rs"
+doctest = false
+
+[dependencies]
+smol.workspace = true
+workspace-hack.workspace = true
+
+[target.'cfg(target_os = "windows")'.dependencies]
+anyhow.workspace = true
+async-io = "2.4"
+windows.workspace = true
+
+[dev-dependencies]
+tempfile.workspace = true

crates/net/LICENSE-GPL

@@ -0,0 +1 @@
+../../LICENSE-GPL

crates/net/src/async_net.rs

@@ -0,0 +1,69 @@
+#[cfg(not(target_os = "windows"))]
+pub use smol::net::unix::{UnixListener, UnixStream};
+
+#[cfg(target_os = "windows")]
+pub use windows::{UnixListener, UnixStream};
+
+#[cfg(target_os = "windows")]
+pub mod windows {
+    use std::{
+        io::Result,
+        path::Path,
+        pin::Pin,
+        task::{Context, Poll},
+    };
+
+    use smol::{
+        Async,
+        io::{AsyncRead, AsyncWrite},
+    };
+
+    pub struct UnixListener(Async<crate::UnixListener>);
+
+    impl UnixListener {
+        pub fn bind<P: AsRef<Path>>(path: P) -> Result<Self> {
+            Ok(UnixListener(Async::new(crate::UnixListener::bind(path)?)?))
+        }
+
+        pub async fn accept(&self) -> Result<(UnixStream, ())> {
+            let (sock, _) = self.0.read_with(|listener| listener.accept()).await?;
+            Ok((UnixStream(Async::new(sock)?), ()))
+        }
+    }
+
+    pub struct UnixStream(Async<crate::UnixStream>);
+
+    impl UnixStream {
+        pub async fn connect<P: AsRef<Path>>(path: P) -> Result<Self> {
+            Ok(UnixStream(Async::new(crate::UnixStream::connect(path)?)?))
+        }
+    }
+
+    impl AsyncRead for UnixStream {
+        fn poll_read(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            buf: &mut [u8],
+        ) -> Poll<Result<usize>> {
+            Pin::new(&mut self.0).poll_read(cx, buf)
+        }
+    }
+
+    impl AsyncWrite for UnixStream {
+        fn poll_write(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<Result<usize>> {
+            Pin::new(&mut self.0).poll_write(cx, buf)
+        }
+
+        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+            Pin::new(&mut self.0).poll_flush(cx)
+        }
+
+        fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+            Pin::new(&mut self.0).poll_close(cx)
+        }
+    }
+}

crates/net/src/listener.rs

@@ -0,0 +1,45 @@
+use std::{
+    io::Result,
+    os::windows::io::{AsSocket, BorrowedSocket},
+    path::Path,
+};
+
+use windows::Win32::Networking::WinSock::{SOCKADDR_UN, SOMAXCONN, bind, listen};
+
+use crate::{
+    socket::UnixSocket,
+    stream::UnixStream,
+    util::{init, map_ret, sockaddr_un},
+};
+
+pub struct UnixListener(UnixSocket);
+
+impl UnixListener {
+    pub fn bind<P: AsRef<Path>>(path: P) -> Result<Self> {
+        init();
+        let socket = UnixSocket::new()?;
+        let (addr, len) = sockaddr_un(path)?;
+        unsafe {
+            map_ret(bind(
+                socket.as_raw(),
+                &addr as *const _ as *const _,
+                len as i32,
+            ))?;
+            map_ret(listen(socket.as_raw(), SOMAXCONN as _))?;
+        }
+        Ok(Self(socket))
+    }
+
+    pub fn accept(&self) -> Result<(UnixStream, ())> {
+        let mut storage = SOCKADDR_UN::default();
+        let mut len = std::mem::size_of_val(&storage) as i32;
+        let raw = self.0.accept(&mut storage as *mut _ as *mut _, &mut len)?;
+        Ok((UnixStream::new(raw), ()))
+    }
+}
+
+impl AsSocket for UnixListener {
+    fn as_socket(&self) -> BorrowedSocket<'_> {
+        unsafe { BorrowedSocket::borrow_raw(self.0.as_raw().0 as _) }
+    }
+}

crates/net/src/net.rs

@@ -0,0 +1,107 @@
+pub mod async_net;
+#[cfg(target_os = "windows")]
+pub mod listener;
+#[cfg(target_os = "windows")]
+pub mod socket;
+#[cfg(target_os = "windows")]
+pub mod stream;
+#[cfg(target_os = "windows")]
+mod util;
+
+#[cfg(target_os = "windows")]
+pub use listener::*;
+#[cfg(target_os = "windows")]
+pub use socket::*;
+#[cfg(not(target_os = "windows"))]
+pub use std::os::unix::net::{UnixListener, UnixStream};
+#[cfg(target_os = "windows")]
+pub use stream::*;
+
+#[cfg(test)]
+mod tests {
+    use std::io::{Read, Write};
+
+    use smol::io::{AsyncReadExt, AsyncWriteExt};
+
+    const SERVER_MESSAGE: &str = "Connection closed";
+    const CLIENT_MESSAGE: &str = "Hello, server!";
+    const BUFFER_SIZE: usize = 32;
+
+    #[test]
+    fn test_windows_listener() -> std::io::Result<()> {
+        use crate::{UnixListener, UnixStream};
+
+        let temp = tempfile::tempdir()?;
+        let socket = temp.path().join("socket.sock");
+        let listener = UnixListener::bind(&socket)?;
+
+        // Server
+        let server = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().unwrap();
+
+            // Read data from the client
+            let mut buffer = [0; BUFFER_SIZE];
+            let bytes_read = stream.read(&mut buffer).unwrap();
+            let string = String::from_utf8_lossy(&buffer[..bytes_read]);
+            assert_eq!(string, CLIENT_MESSAGE);
+
+            // Send a message back to the client
+            stream.write_all(SERVER_MESSAGE.as_bytes()).unwrap();
+        });
+
+        // Client
+        let mut client = UnixStream::connect(&socket)?;
+
+        // Send data to the server
+        client.write_all(CLIENT_MESSAGE.as_bytes())?;
+        let mut buffer = [0; BUFFER_SIZE];
+
+        // Read the response from the server
+        let bytes_read = client.read(&mut buffer)?;
+        let string = String::from_utf8_lossy(&buffer[..bytes_read]);
+        assert_eq!(string, SERVER_MESSAGE);
+        client.flush()?;
+
+        server.join().unwrap();
+        Ok(())
+    }
+
+    #[test]
+    fn test_unix_listener() -> std::io::Result<()> {
+        use crate::async_net::{UnixListener, UnixStream};
+
+        smol::block_on(async {
+            let temp = tempfile::tempdir()?;
+            let socket = temp.path().join("socket.sock");
+            let listener = UnixListener::bind(&socket)?;
+
+            // Server
+            let server = smol::spawn(async move {
+                let (mut stream, _) = listener.accept().await.unwrap();
+
+                // Read data from the client
+                let mut buffer = [0; BUFFER_SIZE];
+                let bytes_read = stream.read(&mut buffer).await.unwrap();
+                let string = String::from_utf8_lossy(&buffer[..bytes_read]);
+                assert_eq!(string, CLIENT_MESSAGE);
+
+                // Send a message back to the client
+                stream.write_all(SERVER_MESSAGE.as_bytes()).await.unwrap();
+            });
+
+            // Client
+            let mut client = UnixStream::connect(&socket).await?;
+            client.write_all(CLIENT_MESSAGE.as_bytes()).await?;
+
+            // Read the response from the server
+            let mut buffer = [0; BUFFER_SIZE];
+            let bytes_read = client.read(&mut buffer).await?;
+            let string = String::from_utf8_lossy(&buffer[..bytes_read]);
+            assert_eq!(string, "Connection closed");
+            client.flush().await?;
+
+            server.await;
+            Ok(())
+        })
+    }
+}

crates/net/src/socket.rs

@@ -0,0 +1,59 @@
+use std::io::{Error, ErrorKind, Result};
+
+use windows::Win32::{
+    Foundation::{HANDLE, HANDLE_FLAG_INHERIT, HANDLE_FLAGS, SetHandleInformation},
+    Networking::WinSock::{
+        AF_UNIX, SEND_RECV_FLAGS, SOCK_STREAM, SOCKADDR, SOCKET, WSA_FLAG_OVERLAPPED,
+        WSAEWOULDBLOCK, WSASocketW, accept, closesocket, recv, send,
+    },
+};
+
+use crate::util::map_ret;
+
+pub struct UnixSocket(SOCKET);
+
+impl UnixSocket {
+    pub fn new() -> Result<Self> {
+        unsafe {
+            let raw = WSASocketW(AF_UNIX as _, SOCK_STREAM.0, 0, None, 0, WSA_FLAG_OVERLAPPED)?;
+            SetHandleInformation(
+                HANDLE(raw.0 as _),
+                HANDLE_FLAG_INHERIT.0,
+                HANDLE_FLAGS::default(),
+            )?;
+            Ok(Self(raw))
+        }
+    }
+
+    pub(crate) fn as_raw(&self) -> SOCKET {
+        self.0
+    }
+
+    pub fn accept(&self, storage: *mut SOCKADDR, len: &mut i32) -> Result<Self> {
+        match unsafe { accept(self.0, Some(storage), Some(len)) } {
+            Ok(sock) => Ok(Self(sock)),
+            Err(err) => {
+                let wsa_err = unsafe { windows::Win32::Networking::WinSock::WSAGetLastError().0 };
+                if wsa_err == WSAEWOULDBLOCK.0 {
+                    Err(Error::new(ErrorKind::WouldBlock, "accept would block"))
+                } else {
+                    Err(err.into())
+                }
+            }
+        }
+    }
+
+    pub(crate) fn recv(&self, buf: &mut [u8]) -> Result<usize> {
+        map_ret(unsafe { recv(self.0, buf, SEND_RECV_FLAGS::default()) })
+    }
+
+    pub(crate) fn send(&self, buf: &[u8]) -> Result<usize> {
+        map_ret(unsafe { send(self.0, buf, SEND_RECV_FLAGS::default()) })
+    }
+}
+
+impl Drop for UnixSocket {
+    fn drop(&mut self) {
+        unsafe { closesocket(self.0) };
+    }
+}

crates/net/src/stream.rs

@@ -0,0 +1,60 @@
+use std::{
+    io::{Read, Result, Write},
+    os::windows::io::{AsSocket, BorrowedSocket},
+    path::Path,
+};
+
+use async_io::IoSafe;
+use windows::Win32::Networking::WinSock::connect;
+
+use crate::{
+    socket::UnixSocket,
+    util::{init, map_ret, sockaddr_un},
+};
+
+pub struct UnixStream(UnixSocket);
+
+unsafe impl IoSafe for UnixStream {}
+
+impl UnixStream {
+    pub fn new(socket: UnixSocket) -> Self {
+        Self(socket)
+    }
+
+    pub fn connect<P: AsRef<Path>>(path: P) -> Result<Self> {
+        init();
+        unsafe {
+            let inner = UnixSocket::new()?;
+            let (addr, len) = sockaddr_un(path)?;
+
+            map_ret(connect(
+                inner.as_raw(),
+                &addr as *const _ as *const _,
+                len as i32,
+            ))?;
+            Ok(Self(inner))
+        }
+    }
+}
+
+impl Read for UnixStream {
+    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
+        self.0.recv(buf)
+    }
+}
+
+impl Write for UnixStream {
+    fn write(&mut self, buf: &[u8]) -> Result<usize> {
+        self.0.send(buf)
+    }
+
+    fn flush(&mut self) -> Result<()> {
+        Ok(())
+    }
+}
+
+impl AsSocket for UnixStream {
+    fn as_socket(&self) -> BorrowedSocket<'_> {
+        unsafe { BorrowedSocket::borrow_raw(self.0.as_raw().0 as _) }
+    }
+}

crates/net/src/util.rs

@@ -0,0 +1,76 @@
+use std::{
+    io::{Error, ErrorKind, Result},
+    path::Path,
+    sync::Once,
+};
+
+use windows::Win32::Networking::WinSock::{
+    ADDRESS_FAMILY, AF_UNIX, SOCKADDR_UN, SOCKET_ERROR, WSAGetLastError, WSAStartup,
+};
+
+pub(crate) fn init() {
+    static ONCE: Once = Once::new();
+
+    ONCE.call_once(|| unsafe {
+        let mut wsa_data = std::mem::zeroed();
+        let result = WSAStartup(0x202, &mut wsa_data);
+        if result != 0 {
+            panic!("WSAStartup failed: {}", result);
+        }
+    });
+}
+
+// https://devblogs.microsoft.com/commandline/af_unix-comes-to-windows/
+pub(crate) fn sockaddr_un<P: AsRef<Path>>(path: P) -> Result<(SOCKADDR_UN, usize)> {
+    let mut addr = SOCKADDR_UN::default();
+    addr.sun_family = ADDRESS_FAMILY(AF_UNIX);
+
+    let bytes = path
+        .as_ref()
+        .to_str()
+        .map(|s| s.as_bytes())
+        .ok_or(ErrorKind::InvalidInput)?;
+
+    if bytes.contains(&0) {
+        return Err(Error::new(
+            ErrorKind::InvalidInput,
+            "paths may not contain interior null bytes",
+        ));
+    }
+    if bytes.len() >= addr.sun_path.len() {
+        return Err(Error::new(
+            ErrorKind::InvalidInput,
+            "path must be shorter than SUN_LEN",
+        ));
+    }
+
+    unsafe {
+        std::ptr::copy_nonoverlapping(
+            bytes.as_ptr(),
+            addr.sun_path.as_mut_ptr().cast(),
+            bytes.len(),
+        );
+    }
+
+    let mut len = sun_path_offset(&addr) + bytes.len();
+    match bytes.first() {
+        Some(&0) | None => {}
+        Some(_) => len += 1,
+    }
+    Ok((addr, len))
+}
+
+pub(crate) fn map_ret(ret: i32) -> Result<usize> {
+    if ret == SOCKET_ERROR {
+        Err(Error::from_raw_os_error(unsafe { WSAGetLastError().0 }))
+    } else {
+        Ok(ret as usize)
+    }
+}
+
+fn sun_path_offset(addr: &SOCKADDR_UN) -> usize {
+    // Work with an actual instance of the type since using a null pointer is UB
+    let base = addr as *const _ as usize;
+    let path = &addr.sun_path as *const _ as usize;
+    path - base
+}